Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ jobs:
runs-on: ubuntu-latest
strategy:
matrix:
node-version: [20.x, 22.x]
node-version: [20.x, 22.x, 24.x]
steps:
- uses: actions/checkout@v4
- name: Use Node.js ${{ matrix.node-version }}
Expand Down
33 changes: 23 additions & 10 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,10 +6,11 @@
[![npm](https://img.shields.io/badge/node-%3E%3D20-brightgreen)](package.json)
[![license](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE)

> Ship software with **many focused AI workers** instead of one overloaded chat thread.
> Session-management MVP for named model conversations and host-verified checks. The live adapter generates text; it does not edit code or execute tools.

```bash
npm install -g workstream # or: npx / local clone
git clone https://github.com/acephos/workstream
cd workstream && npm ci && npm run build && npm link
ws init my-project
ws spawn --name auth --role good --prompt "Design JWT refresh flow"
ws spawn --name ui --role fast --prompt "Scaffold login form components"
Expand All @@ -31,9 +32,9 @@ Single-thread agent chats collapse under real product work:
| “Done?” is vibes | Optional **delivery contracts** (`ws check`) |
| Hard to parallelize | Named sessions you can `send` / `wait` / `list` |

**workstream** is a small, local-first CLI + library that models how agent product teams actually ship: spawn workers, route by role, keep durable turn logs, and validate “done” against explicit criteria.
**workstream** is a local CLI and library for named model conversations, role routing, durable turn logs, and explicit delivery criteria.

It is a **public demonstration** of agent orchestration patterns used when shipping software with AI workers. Built by [Aniket Singh](https://github.com/acephos) while working on agentic product tooling in industry. **Not affiliated with any employer; contains no proprietary code.**
Built by [Aniket Singh](https://github.com/acephos) as a public experiment in model-session management.

---

Expand Down Expand Up @@ -80,12 +81,12 @@ ws done codemod

### Delivery contracts

Attach a criteria file when spawning. `ws check` validates it.
Attach a criteria file when spawning. `ws check --run-checks` executes explicit JSON argv commands and writes host receipts. Without this flag commands remain unverified. `file:` checks record artifact hashes and only prove presence. Transcript text never proves completion. Missing/empty criteria fail closed.

```text
# criteria/auth.txt
# one criterion per line; # comments ok
text:tests pass
command:["npm","test"]
file:dist/cli.js
```

Expand All @@ -95,7 +96,7 @@ ws spawn --name ship --role good \
--criteria criteria/auth.txt

ws send ship "CI green — tests pass"
ws check ship # FAIL until file:dist/cli.js exists, then PASS
ws check ship --run-checks # executes the configured host check; failure or missing artifact rejects completion
```

---
Expand Down Expand Up @@ -210,7 +211,7 @@ const results = await harness.check();
1. **Sessions are first-class** — not hidden threads inside one blob of chat history.
2. **Filesystem is the source of truth** — inspectable JSON; no daemon required for MVP.
3. **Adapters are swappable** — mock for CI; OpenAI-compatible HTTP for real models.
4. **Delivery contracts are optional but real** — `file:` and `text:` criteria turn “done” into something checkable.
4. **Delivery contracts are optional but real** — `file:` and `command:` criteria turn “done” into something checkable.
5. **Clean-room public code** — original implementation for portfolio / open collaboration.

---
Expand Down Expand Up @@ -244,7 +245,7 @@ npm run typecheck
npm run build
```

CI runs on push/PR via GitHub Actions (Node 20 & 22).
CI runs on push/PR via GitHub Actions (Node 20, 22, and 24).

---

Expand All @@ -262,10 +263,22 @@ CI runs on push/PR via GitHub Actions (Node 20 & 22).

**Aniket Singh** — [github.com/acephos](https://github.com/acephos)

Built as a public demonstration of multi-agent orchestration patterns used when shipping software with AI workers. Written while working on agentic product tooling in industry. **Not affiliated with NCR Voyix or any employer; no company code or proprietary IP.**
Public experiment in named model sessions and host-verified delivery criteria.

---

## License

[MIT](LICENSE) © 2026 Aniket Singh

## Reliability and verification boundaries

All session mutations are serialized with a local filesystem lock across CLI processes and Harness instances. Atomic snapshots use unique temporary files. A second send waits for the first to finish; separate sessions can progress independently. Locks do not support shared network filesystems. After an interrupted process, an abandoned lock fails visibly: verify no writer is active before removing the named `.lock` directory. Corrupt session files are surfaced instead of silently omitted.

The live HTTP adapter times out after 60 seconds. Library callers can pass `{ signal }` to `harness.send(name, message, { signal })` to cancel a request; failure persists the user turn and error. Injected adapters must honor that signal themselves.

Delivery criteria accept `file:relative/path` and `command:["executable","argument"]`. Commands never run through a shell and require `--run-checks` (or `{ runCommands: true }` in the library). Review a criteria file before enabling host commands. Checks time out after 60 seconds and store exit status, command arguments, timestamps, criteria hash, and output hash under `.workstream/evidence/`. File criteria must resolve within the workspace. Commands prove the behavior they actually test; a passing suite is not exhaustive coverage. Deprecated `text:` and bare transcript criteria fail with migration guidance.

## Contributing

Use a source checkout and run `npm ci`, `npm test`, and `npm run typecheck`. CI checks Node 20, 22, and 24. Changes to persistence or delivery checks should include concurrent-process, failure, and denial cases. Registry distribution is not assumed by this README; use the source-install instructions above.
4 changes: 2 additions & 2 deletions package.json
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
{
"name": "workstream",
"version": "0.1.0",
"description": "Multi-session agent orchestration harness — parallel workstreams, role routing, delivery contracts.",
"description": "Multi-session agent orchestration harness \u2014 parallel workstreams, role routing, delivery contracts.",
"type": "module",
"bin": {
"workstream": "./dist/cli.js",
Expand All @@ -23,7 +23,7 @@
"scripts": {
"build": "tsc -p tsconfig.json",
"prepare": "npm run build",
"test": "npm run build && node --test --test-reporter=spec dist/tests/harness.test.js dist/tests/cli.test.js",
"test": "npm run build && node --test --test-reporter=spec dist/tests/*.test.js",
"cli": "node dist/cli.js",
"typecheck": "tsc -p tsconfig.json --noEmit"
},
Expand Down
4 changes: 4 additions & 0 deletions src/adapters/openai.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,11 @@ export class OpenAIAdapter implements Adapter {
private readonly apiKey: string;
private readonly baseUrl: string;
private readonly model: string;
private readonly timeoutMs: number;

constructor(options: AdapterFactoryOptions = {}) {
this.timeoutMs = options.timeoutMs ?? 60_000;
if (!Number.isFinite(this.timeoutMs) || this.timeoutMs <= 0) throw new Error("timeoutMs must be positive");
const apiKey = options.apiKey ?? process.env.WORKSTREAM_API_KEY;
if (!apiKey) {
throw new Error(
Expand All @@ -35,6 +38,7 @@ export class OpenAIAdapter implements Adapter {
const url = `${this.baseUrl}/chat/completions`;

const res = await fetch(url, {
signal: request.signal ? AbortSignal.any([request.signal, AbortSignal.timeout(this.timeoutMs)]) : AbortSignal.timeout(this.timeoutMs),
method: "POST",
headers: {
"content-type": "application/json",
Expand Down
1 change: 1 addition & 0 deletions src/adapters/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,4 +10,5 @@ export interface AdapterFactoryOptions {
model?: string;
apiKey?: string;
baseUrl?: string;
timeoutMs?: number;
}
156 changes: 66 additions & 90 deletions src/check.ts
Original file line number Diff line number Diff line change
@@ -1,103 +1,79 @@
import { readFile, access } from "node:fs/promises";
import { readFile, stat, realpath } from "node:fs/promises";
import { spawn } from "node:child_process";
import { createHash, randomUUID } from "node:crypto";
import path from "node:path";
import type { CheckResult, Session } from "./types.js";
import { writeJsonAtomic, resolveWorkstreamRoot } from "./storage.js";

/**
* Delivery contract: optional criteria file listing required outcomes.
*
* Format (plain text, one criterion per line):
* - Lines starting with # are comments
* - Empty lines ignored
* - Optional prefix: `file:` path that must exist relative to workspace
* - Optional prefix: `text:` substring that must appear in session turns
* - Bare lines are treated as `text:` criteria
*
* Example criteria file:
* # Ship checklist
* file:dist/cli.js
* text:tests pass
* done criteria documented
*/
export async function checkDeliveryContract(
session: Session,
cwd: string,
): Promise<CheckResult> {
const notes: string[] = [];
const missing: string[] = [];

if (!session.criteriaFile) {
notes.push("No criteriaFile set on session; nothing to validate.");
return {
session: session.name,
ok: true,
missing,
notes,
};
}

const criteriaPath = path.resolve(cwd, session.criteriaFile);
try {
await access(criteriaPath);
} catch {
return {
session: session.name,
ok: false,
criteriaFile: session.criteriaFile,
missing: [`criteria file missing: ${session.criteriaFile}`],
notes: ["Attach a criteria file with `ws spawn --criteria <path>`."],
};
}

const raw = await readFile(criteriaPath, "utf8");
const lines = raw
.split(/\r?\n/)
.map((l) => l.trim())
.filter((l) => l.length > 0 && !l.startsWith("#"));

if (lines.length === 0) {
notes.push("Criteria file is empty (only comments/blank lines).");
return {
session: session.name,
ok: true,
criteriaFile: session.criteriaFile,
missing,
notes,
};
async function workspacePath(cwd: string, name: string): Promise<string> {
const root = await realpath(cwd);
const target = await realpath(path.resolve(cwd, name));
const relative = path.relative(root, target);
if (relative === ".." || relative.startsWith(`..${path.sep}`) || path.isAbsolute(relative)) {
throw new Error("Evidence must be inside the workspace");
}
return target;
}

const transcript = session.turns.map((t) => t.content).join("\n");
async function runCheck(argv: string[], cwd: string, timeoutMs: number) {
const startedAt = new Date().toISOString();
const digest = createHash("sha256");
let error: string | undefined;
let timedOut = false;
const exitCode = await new Promise<number | null>((resolve) => {
const child = spawn(argv[0]!, argv.slice(1), { cwd, shell: false, detached: process.platform !== "win32", stdio: ["ignore", "pipe", "pipe"] });
child.stdout.on("data", chunk => digest.update(chunk));
child.stderr.on("data", chunk => digest.update(chunk));
const timer = setTimeout(() => {
timedOut = true;
if (process.platform === "win32" && child.pid) {
spawn("taskkill", ["/pid", String(child.pid), "/t", "/f"], { stdio: "ignore" });
} else if (child.pid) {
try { process.kill(-child.pid, "SIGKILL"); } catch { child.kill("SIGKILL"); }
}
}, timeoutMs);
child.on("error", err => { error = err.message; });
child.on("close", code => { clearTimeout(timer); resolve(code); });
});
return { argv, startedAt, finishedAt: new Date().toISOString(), exitCode, timedOut, error, outputSha256: digest.digest("hex"), ok: exitCode === 0 && !timedOut && !error };
}

/** Evidence checks: file hashes or explicitly enabled host commands, never transcript claims. */
export async function checkDeliveryContract(session: Session, cwd: string, options: { runCommands?: boolean; timeoutMs?: number } = {}): Promise<CheckResult> {
const notes: string[] = [];
const missing: string[] = [];
const result = () => ({ session: session.name, ok: missing.length === 0, criteriaFile: session.criteriaFile, missing, notes });
if (!session.criteriaFile) { missing.push("No criteria file; completion is unverified."); return result(); }
let raw: string;
try { raw = await readFile(await workspacePath(cwd, session.criteriaFile), "utf8"); }
catch { missing.push(`criteria file missing or outside workspace: ${session.criteriaFile}`); return result(); }
const lines = raw.split(/\r?\n/).map(line => line.trim()).filter(line => line && !line.startsWith("#"));
if (!lines.length) { missing.push("Empty criteria file; completion is unverified."); return result(); }
const timeoutMs = options.timeoutMs ?? 60_000;
if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) throw new Error("Check timeout must be positive");
for (const line of lines) {
if (line.startsWith("file:")) {
const rel = line.slice("file:".length).trim();
const target = path.resolve(cwd, rel);
const name = line.slice(5).trim();
try {
await access(target);
notes.push(`present: file:${rel}`);
} catch {
missing.push(`file:${rel}`);
}
continue;
}

const needle = line.startsWith("text:")
? line.slice("text:".length).trim()
: line;

if (needle.length === 0) continue;

if (transcript.includes(needle)) {
notes.push(`present: text:${needle}`);
const target = await workspacePath(cwd, name);
if (!(await stat(target)).isFile()) throw new Error("not a file");
const hash = createHash("sha256").update(await readFile(target)).digest("hex");
notes.push(`present: file:${name} sha256=${hash} (artifact presence, not behavior verification)`);
} catch { missing.push(line); }
} else if (line.startsWith("command:")) {
try {
const argv: unknown = JSON.parse(line.slice(8));
if (!Array.isArray(argv) || !argv.length || !argv.every(x => typeof x === "string" && x.length > 0)) throw new Error("Expected a nonempty JSON argv array");
if (!options.runCommands) { missing.push(`${line} (requires --run-checks)`); continue; }
const receipt = await runCheck(argv, cwd, timeoutMs);
const receiptPath = path.join(resolveWorkstreamRoot(cwd), "evidence", `${session.name}-${randomUUID()}.json`);
await writeJsonAtomic(receiptPath, { session: session.name, criteriaSha256: createHash("sha256").update(raw).digest("hex"), ...receipt });
notes.push(`host check receipt: ${path.relative(cwd, receiptPath)}`);
if (!receipt.ok) missing.push(`${line} (exit=${receipt.exitCode}, timedOut=${receipt.timedOut}${receipt.error ? `, error=${receipt.error}` : ""})`);
} catch (error) { missing.push(`${line}: ${error instanceof Error ? error.message : String(error)}`); }
} else {
missing.push(`text:${needle}`);
missing.push(`${line}: transcript claims cannot verify completion; use file: or command:["executable","arg"]`);
}
}

return {
session: session.name,
ok: missing.length === 0,
criteriaFile: session.criteriaFile,
missing,
notes,
};
return result();
}
4 changes: 2 additions & 2 deletions src/cli.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ Usage:
ws list
ws wait <name> [--timeout <ms>] [--mark-only]
ws status
ws check [name]
ws check [name] [--run-checks]
ws done <name>
ws help
ws version
Expand Down Expand Up @@ -174,7 +174,7 @@ async function main(argv: string[]): Promise<number> {
}
case "check": {
const name = positionals[0];
const results = await harness.check(name);
const results = await harness.check(name, { runCommands: flagBool(flags, "run-checks") });
if (results.length === 0) {
console.log("(no sessions to check)");
return 0;
Expand Down
Loading
Loading