From 5c4c13d06a3ef2e32753ca02435b30c585c2442b Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 13 Jul 2026 14:12:18 +0000 Subject: [PATCH 1/2] Add the tracker layer: tickets, cross-task deps, state model, API, real workers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the review of the first PR ("reference impl of the dispatcher/ execution layer, not the tracker layer") — the pieces needed to serve as a Plane replacement rather than just a per-task engine: - Ticket entity + tasks.json registry (§1.5): the #N / #N.x tree, needs cross-task dependencies (acyclic by construction), locked + versioned local registry with atomic writes - Cross-ticket DAG promotion: blockedBy -> promote on done; unmet needs sit in backlog, promotion staffs autoStaff tickets in the background, missed promotions are re-derived on recovery - The eight-value state model (§1.1): backlog/todo/design/design_review/ in_progress/in_review/done/cancelled, projected live from the run record via a per-ticket state_map (INT map shipped as a preset), with park reasons surfaced on the ticket - HTTP API (node:http, dependency-free): file/board/status, staff, nudge, gate verdicts, pause/resume/cancel, events/journal reads; 400/404/409 error mapping; owns the engine tick cadence - CLI is no longer read-only: serve boots the full stack (tracker + git worktrees + process workers); board/file/ticket/staff/nudge/gate/pause/ resume/cancel drive the API - Real agent-spawn adapter: spawns an actual worker process per seat in its git worktree, writes the stage manifest + brief into .beckett/, speaks a line-JSON driver protocol (ready handshake, turns, file changes, checkpoints, done-signal), nudges over stdin, abort = WIP commit + SIGTERM, silent exits synthesize finished and ride the alarm/retry ladder - CI: GitHub Actions (typecheck, tests, simulated deployments) - 26 new tests (151 total): tracker semantics, promotion, projection, registry versioning, real-child-process end to end (incl. under the tracker), full API lifecycle over the wire Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01A3DUAdiNuM62Q6TVAyj7rv --- .github/workflows/ci.yml | 21 ++ README.md | 62 +++++- src/adapters/process.ts | 323 ++++++++++++++++++++++++++++++ src/cli.ts | 176 +++++++++++++++- src/engine/stageManager.ts | 9 + src/index.ts | 7 + src/presets.ts | 13 ++ src/server.ts | 192 ++++++++++++++++++ src/tracker/store.ts | 120 +++++++++++ src/tracker/ticket.ts | 78 ++++++++ src/tracker/tracker.ts | 376 +++++++++++++++++++++++++++++++++++ test/process-adapter.test.ts | 254 +++++++++++++++++++++++ test/server.test.ts | 190 ++++++++++++++++++ test/tracker.test.ts | 283 ++++++++++++++++++++++++++ 14 files changed, 2085 insertions(+), 19 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 src/adapters/process.ts create mode 100644 src/server.ts create mode 100644 src/tracker/store.ts create mode 100644 src/tracker/ticket.ts create mode 100644 src/tracker/tracker.ts create mode 100644 test/process-adapter.test.ts create mode 100644 test/server.test.ts create mode 100644 test/tracker.test.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..8512a17 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,21 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-node@v4 + with: + node-version: 22 + cache: npm + - run: npm ci + - run: npm run typecheck + - run: npm test + - name: simulated deployments (§7) + run: npm run deploy:all diff --git a/README.md b/README.md index 713796c..b80dbc1 100644 --- a/README.md +++ b/README.md @@ -1,12 +1,18 @@ # run-of-show -A per-task **dynamic workflow engine**, built as an embeddable service -primitive. It implements the *Run of Show* design proposal (OPS-151 → OPS-152, -rev 2): instead of one compiled-in ticket lifecycle, every task carries a -small, validated DAG — the **flow spec** — that a concierge writes at filing -time. A generic, event-sourced **stage manager** interprets it: stages, -iteration budgets, parallel fan-out, join rules and human gates are data, not -dispatcher source. +A per-task **dynamic workflow engine with a ticket tracker on top** — both +layers of the *Run of Show* design proposal (OPS-151 → OPS-152, rev 2): + +- **the execution layer** (dispatcher replacement): every task carries a + small, validated DAG — the **flow spec** — interpreted by a generic, + event-sourced **stage manager**. Stages, iteration budgets, parallel + fan-out, join rules and human gates are data, not dispatcher source. +- **the tracker layer** (Plane replacement): a real **ticket entity** on the + `#N / #N.x` tree, **cross-task `needs` dependencies** with + promote-on-done, the **eight-value state model** (backlog / todo / design / + design_review / in_progress / in_review / done / cancelled) projected live + from the run record, an **HTTP API + CLI** to drive it all, and a **real + agent-spawn adapter** that runs actual worker processes in git worktrees. The engine holds the design's two reliability invariants: @@ -52,16 +58,44 @@ src/ engine/stageManager.ts the interpreter: §4.3's append→fold→decide→gate→act→announce adapters/simulated.ts hand-cranked workers + fake worktrees (tests, simulations) adapters/git.ts real git worktrees, branches and merges + adapters/process.ts REAL workers: child processes + the JSONL driver protocol + tracker/ticket.ts the ticket entity: #N/#N.x refs, needs, states (§1.1/§1.5) + tracker/store.ts tasks.json — the local, locked, versioned registry + tracker/tracker.ts filing, staffing, promote-on-done, state projection + server.ts the HTTP API over the tracker presets.ts Appendix A: today's lifecycles as specs, plus §8 shapes - cli.ts lint / status / journal / events + cli.ts lint/status/journal/events + serve/board/file/nudge/gate/… examples/ §7 — the three simulated deployments, runnable -test/ 122+ tests, unit through end-to-end +test/ 150+ tests, unit through end-to-end ``` Storage layout under the engine root (default `~/.beckett` for the CLI): `runs/.jsonl` (the source of truth), `runs/.head.json` (cache only), `journal/.log`, `spend.jsonl`. +## Running it as a tracker (server + real workers) + +```bash +# boot the board over a repo; every seat spawns your worker command in its +# own git worktree, speaking the JSONL driver protocol (adapters/process.ts) +npx run-of-show serve --root ~/.beckett --repo /path/to/repo \ + --worker "node my-worker.cjs" --port 7770 + +# file work over the wire (auto-staffs when its needs are met) +npx run-of-show file --title "notification prefs API" --body "…" --criteria "stores prefs" +npx run-of-show file --title "prefs UI" --needs "#1" # promotes when #1 is done +npx run-of-show board +npx run-of-show gate "#1" --node design_review --verdict pass +npx run-of-show nudge "#1" "prefer the small fix" --node implement +``` + +A worker is any executable: it gets `BECKETT_MANIFEST` (the §6.4 stage +manifest, written into `.beckett/stage-manifest.json` in its worktree) and +`BECKETT_BRIEF`, proves where it is (`worker_ready` with the manifest hash + +the branch/sha it observes via git), emits progress events on stdout, and +finishes with a done-signal. A process that dies silently is alarmed, +retried, and parked by the engine — never lost. + ## Using it as a primitive ```ts @@ -137,6 +171,16 @@ Coverage highlights: - **Appendix A equivalence**: OPS reviewed lifecycle, one-pass, and INT design flow walk the same stages with the same caps as today's constants (`MAX_REWORK_CYCLES`, `MAX_IMPLEMENT_RETRIES`, `MAX_DESIGN_CYCLES`). +- **Tracker layer**: ref allocation and the one-level tree, needs validation, + promote-on-done (single, multiple, and autoStaff:false needs), state + projection across all three lifecycles incl. the INT state_map, tracker + recovery with missed promotions, and the versioned/locked registry. +- **Real workers**: an actual node child process handshakes against real git + state, commits real output and completes the run; silent exits alarm and + re-staff; refusals walk the refusal path; nudges arrive over stdin. +- **HTTP API**: the full lifecycle driven over the wire (file, staff, nudge, + gate verdicts, pause/resume, cancel, deps promotion) with 400/404/409 + error mapping. ## CLI diff --git a/src/adapters/process.ts b/src/adapters/process.ts new file mode 100644 index 0000000..b8eaa4f --- /dev/null +++ b/src/adapters/process.ts @@ -0,0 +1,323 @@ +/** + * The real agent-spawn adapter — the piece the review called "literally + * unimplemented, so it can't run work". It spawns an actual worker process + * per seat (any harness CLI: claude, codex, a shell script) inside the + * seat's git worktree, and speaks a line-oriented JSON driver protocol: + * + * worker stdout → engine (one JSON object per line): + * {"kind":"session_started"} + * {"kind":"worker_ready","manifestHash":…,"observedBranch":…,"observedSha":…} + * {"kind":"worker_refused","observed":{…}} + * {"kind":"turn_completed","turn":N,"toolCalls":N,"tokens":{"input":N,"output":N}} + * {"kind":"file_change","path":…} + * {"kind":"checkpoint","sha":…} + * {"kind":"stalled"} + * {"kind":"finished","signal":{…done-signal…},"spendUsd":N} + * + * engine → worker stdin: + * {"kind":"nudge","text":…} + * + * Before launch the adapter writes the §6.4 stage manifest to + * `.beckett/stage-manifest.json` and the brief to `.beckett/brief.md` in + * the worktree, and exports BECKETT_MANIFEST / BECKETT_BRIEF so the worker + * can complete the readiness handshake (echo the manifest hash plus the + * branch/sha it actually observes via git). + * + * Failure honesty is preserved end to end: non-JSON stdout is ignored as + * chatter; a process that exits without a `finished` line synthesizes + * `finished{signal:null}` — the engine's silent-exit alarm and retry ladder + * take it from there. Abort commits WIP in the worktree, then terminates + * the process tree. + */ +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import * as fs from "node:fs"; +import * as path from "node:path"; +import type { SeatRequest, SpawnAdapter, TokenUsage, WorkerEvent, WorkerHandle } from "../engine/ports.js"; +import type { DoneSignal } from "../run/events.js"; +import { GitWorktreeSpawnAdapter } from "./git.js"; +import type { Deliver } from "./simulated.js"; + +export interface WorkerCommand { + cmd: string; + args: string[]; + env?: Record; +} + +/** Resolve the command for a seat — typically keyed off request.cast.harness. */ +export type CommandResolver = (request: SeatRequest) => WorkerCommand; + +interface LiveProcess { + request: SeatRequest; + child: ChildProcessWithoutNullStreams; + startedAtMs: number; + finishedDelivered: boolean; + aborted: boolean; + turns: number; + toolCalls: number; + tokens: TokenUsage; +} + +export class ProcessSpawnAdapter implements SpawnAdapter { + private deliverFn: Deliver | null = null; + private readonly live = new Map(); + /** per-seat delivery chains keep driver events strictly ordered */ + private readonly chains = new Map>(); + + constructor( + /** provisioning (worktrees, branches, base shas, WIP commits) is git's */ + private readonly git: GitWorktreeSpawnAdapter, + private readonly command: CommandResolver, + private readonly opts: { killGraceMs?: number } = {}, + ) {} + + connect(deliver: Deliver): void { + this.deliverFn = deliver; + } + + provision(id: { + ref: string; + node: string; + visit: number; + arm?: number; + isolation?: "worktree-each" | "shared"; + }): { worktree: string; branch: string; baseSha: string } { + return this.git.provision(id); + } + + baseShaFor(ref: string): string { + return this.git.baseShaFor(ref); + } + + /** Wait for every live child to exit (drain for tests / shutdown). */ + async drain(): Promise { + while (this.live.size > 0 || (await this.anyChainPending())) { + await new Promise((resolve) => setTimeout(resolve, 20)); + } + } + + private async anyChainPending(): Promise { + await Promise.all([...this.chains.values()]); + return false; + } + + spawn(request: SeatRequest): WorkerHandle { + if (!this.deliverFn) throw new Error("ProcessSpawnAdapter not connected to an engine"); + const dotdir = path.join(request.worktree, ".beckett"); + fs.mkdirSync(dotdir, { recursive: true }); + const manifestPath = path.join(dotdir, "stage-manifest.json"); + fs.writeFileSync(manifestPath, JSON.stringify(request.manifest, null, 2)); + const briefPath = path.join(dotdir, "brief.md"); + fs.writeFileSync(briefPath, renderBrief(request)); + + const { cmd, args, env } = this.command(request); + const child = spawn(cmd, args, { + cwd: request.worktree, + env: { + ...process.env, + ...env, + BECKETT_MANIFEST: manifestPath, + BECKETT_BRIEF: briefPath, + BECKETT_SEAT: request.seatKey, + BECKETT_REF: request.ref, + }, + stdio: ["pipe", "pipe", "pipe"], + }); + + const key = `${request.ref}::${request.seatKey}`; + const proc: LiveProcess = { + request, + child, + startedAtMs: Date.now(), + finishedDelivered: false, + aborted: false, + turns: 0, + toolCalls: 0, + tokens: { input: 0, output: 0 }, + }; + this.live.set(key, proc); + + let buffer = ""; + child.stdout.on("data", (chunk: Buffer) => { + buffer += chunk.toString("utf8"); + let idx: number; + while ((idx = buffer.indexOf("\n")) >= 0) { + const line = buffer.slice(0, idx).trim(); + buffer = buffer.slice(idx + 1); + if (line) this.handleLine(key, proc, line); + } + }); + child.stderr.on("data", () => { + /* worker chatter; the done-signal path carries the stderr tail on crash */ + }); + child.on("exit", (code) => { + const tail = buffer.trim(); + if (tail) this.handleLine(key, proc, tail); + if (!proc.finishedDelivered && !proc.aborted) { + // Died without a done-signal — the engine synthesizes and alarms. + this.enqueue(key, proc, { + kind: "finished", + signal: null, + error: `process exited with code ${code} without a done-signal`, + }); + } + this.live.delete(key); + }); + + return { + seatKey: request.seatKey, + nudge: (text: string) => { + if (proc.child.exitCode != null || proc.aborted) return { receipt: "dropped" as const }; + try { + proc.child.stdin.write(JSON.stringify({ kind: "nudge", text }) + "\n"); + return { receipt: "delivered" as const }; + } catch { + return { receipt: "dropped" as const }; + } + }, + abort: async (reason: string) => { + proc.aborted = true; + void reason; + const sha = this.git.commitWip(request.worktree, `WIP: seat aborted`); + proc.child.kill("SIGTERM"); + const grace = this.opts.killGraceMs ?? 3000; + setTimeout(() => { + if (proc.child.exitCode == null) proc.child.kill("SIGKILL"); + }, grace).unref(); + return sha; + }, + telemetry: () => ({ + turns: proc.turns, + toolCalls: proc.toolCalls, + tokens: proc.tokens, + wallClockS: (Date.now() - proc.startedAtMs) / 1000, + }), + }; + } + + private handleLine(key: string, proc: LiveProcess, line: string): void { + let parsed: unknown; + try { + parsed = JSON.parse(line); + } catch { + return; // non-protocol stdout is worker chatter, not an event + } + const ev = coerceWorkerEvent(parsed); + if (!ev) return; + if (ev.kind === "turn_completed") { + proc.turns += 1; + proc.toolCalls += ev.toolCalls; + proc.tokens = { + input: proc.tokens.input + ev.tokens.input, + output: proc.tokens.output + ev.tokens.output, + }; + } + if (ev.kind === "finished") proc.finishedDelivered = true; + this.enqueue(key, proc, ev); + } + + private enqueue(key: string, proc: LiveProcess, ev: WorkerEvent): void { + const prev = this.chains.get(key) ?? Promise.resolve(); + const next = prev.then(() => + this.deliverFn!(proc.request.ref, proc.request.seatKey, ev).catch(() => { + /* engine dropped a late event — fine */ + }), + ); + this.chains.set(key, next); + } +} + +/** Validate a decoded protocol line into a WorkerEvent (or reject it). */ +export function coerceWorkerEvent(data: unknown): WorkerEvent | null { + if (typeof data !== "object" || data === null) return null; + const ev = data as Record; + switch (ev["kind"]) { + case "session_started": + case "stalled": + return { kind: ev["kind"] }; + case "worker_ready": + if ( + typeof ev["manifestHash"] === "string" && + typeof ev["observedBranch"] === "string" && + typeof ev["observedSha"] === "string" + ) { + return { + kind: "worker_ready", + manifestHash: ev["manifestHash"], + observedBranch: ev["observedBranch"], + observedSha: ev["observedSha"], + }; + } + return null; + case "worker_refused": + return { + kind: "worker_refused", + observed: (ev["observed"] as Record) ?? {}, + }; + case "turn_completed": { + const tokens = ev["tokens"] as { input?: number; output?: number } | undefined; + return { + kind: "turn_completed", + turn: Number(ev["turn"] ?? 0), + toolCalls: Number(ev["toolCalls"] ?? 0), + tokens: { input: Number(tokens?.input ?? 0), output: Number(tokens?.output ?? 0) }, + }; + } + case "file_change": + return typeof ev["path"] === "string" ? { kind: "file_change", path: ev["path"] } : null; + case "checkpoint": + return typeof ev["sha"] === "string" ? { kind: "checkpoint", sha: ev["sha"] } : null; + case "finished": { + const raw = ev["signal"]; + let signal: DoneSignal | null = null; + if (raw && typeof raw === "object") { + const s = raw as Record; + if (["complete", "blocked", "partial"].includes(s["status"] as string)) { + signal = { + status: s["status"] as DoneSignal["status"], + summary: String(s["summary"] ?? ""), + filesChanged: Array.isArray(s["filesChanged"]) ? (s["filesChanged"] as string[]) : [], + checksRun: Array.isArray(s["checksRun"]) ? (s["checksRun"] as string[]) : null, + blockedReason: typeof s["blockedReason"] === "string" ? s["blockedReason"] : null, + ...(s["data"] && typeof s["data"] === "object" + ? { data: s["data"] as Record } + : {}), + }; + } + } + return { + kind: "finished", + signal, + ...(typeof ev["error"] === "string" ? { error: ev["error"] } : {}), + ...(typeof ev["spendUsd"] === "number" ? { spendUsd: ev["spendUsd"] } : {}), + }; + } + default: + return null; + } +} + +function renderBrief(request: SeatRequest): string { + const b = request.briefParts; + const lines = [ + `# ${request.ref} — ${request.node} (visit ${request.visit}${request.arm != null ? `, arm ${request.arm}` : ""})`, + "", + b.body, + ]; + if (b.criteria.length) { + lines.push("", "## Acceptance criteria", ...b.criteria.map((c) => `- ${c}`)); + } + if (b.nodeBrief) lines.push("", "## Stage instruction", b.nodeBrief); + if (b.rubric) lines.push("", "## Rubric", b.rubric); + if (b.priorArtifacts.length) { + lines.push("", "## Prior artifacts", ...b.priorArtifacts.map((a) => `- ${a.path} (from ${a.fromNode})`)); + } + if (b.steers.length) { + lines.push("", "## Steering", ...b.steers.map((s) => `- ${s.text}`)); + } + lines.push( + "", + "## Envelope (advisory)", + `- ${request.envelope.turnCap} turns / ${request.envelope.wallClockS}s`, + ); + return lines.join("\n") + "\n"; +} diff --git a/src/cli.ts b/src/cli.ts index c4907d8..2df3794 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -1,13 +1,27 @@ #!/usr/bin/env node /** - * A minimal operator surface: lint a spec, render a run's status DAG, - * tail a journal, dump the event log — "which is more visibility than the - * Plane board offers now, not less" (§5.5). + * The operator surface. Read verbs go straight to the store; mutation goes + * through the HTTP API (or boots the whole stack with `serve`), so the CLI + * is no longer read-only. * + * local reads: * run-of-show lint [--max-workers N] * run-of-show status [--root DIR] * run-of-show journal [--root DIR] [--tail N] * run-of-show events [--root DIR] [--tail N] + * + * the server: + * run-of-show serve [--root DIR] [--repo DIR] [--port N] [--worker "cmd arg…"] + * + * against a server (--api http://127.0.0.1:PORT): + * run-of-show board + * run-of-show file --title T [--body B] [--criteria C]… [--needs R]… + * [--parent R] [--flow spec.json] [--no-auto-staff] + * run-of-show ticket + * run-of-show staff + * run-of-show nudge "text" [--node N] + * run-of-show gate --node N --verdict pass|fail [--note T] + * run-of-show pause | resume [--extra-visits N] | cancel [--reason T] */ import * as fs from "node:fs"; import * as path from "node:path"; @@ -15,22 +29,59 @@ import * as os from "node:os"; import { lintFlowSpec } from "./spec/lint.js"; import { maxVisitsOf } from "./spec/types.js"; import { RunStore } from "./run/store.js"; +import { Tracker } from "./tracker/tracker.js"; +import { TrackerServer } from "./server.js"; +import { GitWorktreeSpawnAdapter, GitMergeProvider } from "./adapters/git.js"; +import { ProcessSpawnAdapter } from "./adapters/process.js"; +import { systemClock, type Announcement } from "./engine/ports.js"; + +const argv = process.argv.slice(2); function arg(flag: string): string | undefined { - const i = process.argv.indexOf(flag); - return i >= 0 ? process.argv[i + 1] : undefined; + const i = argv.indexOf(flag); + return i >= 0 ? argv[i + 1] : undefined; +} + +function args(flag: string): string[] { + const out: string[] = []; + for (let i = 0; i < argv.length; i++) { + if (argv[i] === flag && argv[i + 1] != null) out.push(argv[i + 1]!); + } + return out; +} + +function has(flag: string): boolean { + return argv.includes(flag); } function rootDir(): string { return arg("--root") ?? path.join(os.homedir(), ".beckett"); } +function apiBase(): string { + return arg("--api") ?? process.env["RUN_OF_SHOW_API"] ?? "http://127.0.0.1:7770"; +} + function fail(msg: string): never { console.error(msg); process.exit(1); } -const [, , command, target] = process.argv; +async function api(method: string, pathname: string, body?: unknown): Promise { + const res = await fetch(`${apiBase()}${pathname}`, { + method, + headers: { "content-type": "application/json" }, + ...(body !== undefined ? { body: JSON.stringify(body) } : {}), + }); + const data = (await res.json()) as { error?: string }; + if (!res.ok) fail(`API ${res.status}: ${data.error ?? "unknown error"}`); + return data; +} + +const refPath = (ref: string, action?: string) => + `/tickets/${encodeURIComponent(ref)}${action ? `/${action}` : ""}`; + +const [command, target] = argv; switch (command) { case "lint": { @@ -40,9 +91,7 @@ switch (command) { const result = lintFlowSpec(data, maxWorkers ? { maxWorkers: Number(maxWorkers) } : {}); if (result.ok) { console.log(`OK — ${Object.keys(result.spec!.nodes).length} nodes, entry "${result.spec!.entry}"`); - if (result.requiresConfirm) { - console.log("note: a seat requires the confirm-before-cast handshake"); - } + if (result.requiresConfirm) console.log("note: a seat requires the confirm-before-cast handshake"); process.exit(0); } for (const issue of result.errors) { @@ -99,6 +148,113 @@ switch (command) { for (const ev of events.slice(-tail)) console.log(JSON.stringify(ev)); break; } + case "serve": { + const root = rootDir(); + const repo = arg("--repo") ?? process.cwd(); + const port = Number(arg("--port") ?? 7770); + const workerCmd = arg("--worker"); + if (!workerCmd) { + fail( + 'serve needs --worker "cmd arg…" — the command spawned per seat in its worktree (speaks the JSONL driver protocol; see src/adapters/process.ts)', + ); + } + const git = new GitWorktreeSpawnAdapter(repo, systemClock); + const [cmd, ...cmdArgs] = workerCmd.split(/\s+/); + const spawner = new ProcessSpawnAdapter(git, () => ({ cmd: cmd!, args: cmdArgs })); + const sink = { + deliver: (a: Announcement) => + console.log(`[${a.severity.toUpperCase()}] → ${a.target}: ${a.text}`), + }; + const tracker = new Tracker(root, spawner, new GitMergeProvider(git), sink, { + clock: systemClock, + maxWorkers: Number(arg("--max-workers") ?? 4), + ownerDM: arg("--owner-dm") ?? "owner", + }); + spawner.connect((ref, seatKey, ev) => tracker.engine.deliverWorkerEvent(ref, seatKey, ev)); + void tracker.recover().then(async () => { + const server = new TrackerServer(tracker); + const bound = await server.listen(port); + console.log(`run-of-show tracker listening on http://127.0.0.1:${bound} (root ${root}, repo ${repo})`); + }); + break; + } + case "board": { + void api("GET", "/tickets").then((data) => { + const tickets = (data as { tickets: Array> }).tickets; + for (const t of tickets) { + console.log( + `${String(t["ref"]).padEnd(8)} ${String(t["state"]).padEnd(13)} ${t["title"]}${t["stateReason"] ? ` (${t["stateReason"]})` : ""}`, + ); + } + }); + break; + } + case "file": { + const title = arg("--title"); + if (!title) fail("usage: run-of-show file --title T [--body B] [--criteria C]… [--needs R]…"); + const flowFile = arg("--flow"); + void api("POST", "/tickets", { + title, + ...(arg("--body") !== undefined ? { body: arg("--body") } : {}), + ...(args("--criteria").length ? { criteria: args("--criteria") } : {}), + ...(args("--needs").length ? { needs: args("--needs") } : {}), + ...(arg("--parent") !== undefined ? { parent: arg("--parent") } : {}), + ...(arg("--channel") !== undefined ? { originChannel: arg("--channel") } : {}), + ...(flowFile ? { flow: JSON.parse(fs.readFileSync(flowFile, "utf8")) } : {}), + ...(has("--no-auto-staff") ? { autoStaff: false } : {}), + }).then((data) => console.log(JSON.stringify((data as { ticket: unknown }).ticket, null, 2))); + break; + } + case "ticket": { + if (!target) fail("usage: run-of-show ticket "); + void api("GET", refPath(target)).then((d) => console.log(JSON.stringify(d, null, 2))); + break; + } + case "staff": { + if (!target) fail("usage: run-of-show staff "); + void api("POST", refPath(target, "staff")).then((d) => console.log(JSON.stringify(d, null, 2))); + break; + } + case "nudge": { + const text = argv[2]; + if (!target || !text) fail('usage: run-of-show nudge "text" [--node N]'); + void api("POST", refPath(target, "nudge"), { + text, + ...(arg("--node") !== undefined ? { node: arg("--node") } : {}), + }).then((d) => console.log(JSON.stringify(d))); + break; + } + case "gate": { + const node = arg("--node"); + const verdict = arg("--verdict"); + if (!target || !node || !verdict) fail("usage: run-of-show gate --node N --verdict pass|fail"); + void api("POST", refPath(target, "gate"), { + node, + verdict, + ...(arg("--note") !== undefined ? { note: arg("--note") } : {}), + }).then((d) => console.log(JSON.stringify(d, null, 2))); + break; + } + case "pause": + case "cancel": { + if (!target) fail(`usage: run-of-show ${command} `); + void api("POST", refPath(target, command), { + ...(arg("--reason") !== undefined ? { reason: arg("--reason") } : {}), + }).then((d) => console.log(JSON.stringify(d, null, 2))); + break; + } + case "resume": { + if (!target) fail("usage: run-of-show resume [--extra-visits N] [--extra-usd N]"); + const grant: Record = {}; + if (arg("--extra-visits")) grant["extraVisits"] = Number(arg("--extra-visits")); + if (arg("--extra-usd")) grant["extraUsd"] = Number(arg("--extra-usd")); + void api("POST", refPath(target, "resume"), Object.keys(grant).length ? { grant } : {}).then( + (d) => console.log(JSON.stringify(d, null, 2)), + ); + break; + } default: - fail("usage: run-of-show ..."); + fail( + "usage: run-of-show ...", + ); } diff --git a/src/engine/stageManager.ts b/src/engine/stageManager.ts index 1320a92..712542d 100644 --- a/src/engine/stageManager.ts +++ b/src/engine/stageManager.ts @@ -90,6 +90,12 @@ export interface StageManagerOptions { blockedModels?: string[]; ownerDM?: string; defaultChannel?: string; + /** + * Programmatic tap on the event stream, invoked synchronously after each + * append + announce. The tracker layer projects ticket states and drives + * cross-task promotion from this. + */ + onEvent?: (ref: string, event: RunEvent, run: FlowRun) => void; } export class StageManager { @@ -99,6 +105,7 @@ export class StageManager { readonly herald: Herald; private readonly clock: Clock; private readonly lintOpts: LintOptions; + private readonly onEvent: StageManagerOptions["onEvent"]; /** live worker handles, keyed ref::seatKey — in-memory only, rebuilt by recovery */ private seats = new Map(); /** runs mid-pause/cancel/fail-fast: signals append but do not advance */ @@ -112,6 +119,7 @@ export class StageManager { opts: StageManagerOptions, ) { this.clock = opts.clock; + this.onEvent = opts.onEvent; this.store = new RunStore(root); this.scheduler = new SeatScheduler(opts.maxWorkers ?? 8); this.sentinel = new Sentinel(opts.hardCapS ?? 3600); @@ -140,6 +148,7 @@ export class StageManager { const event = this.store.append(ref, this.now(), input); const run = this.store.fold(ref); this.herald.announce(ref, event, run); + this.onEvent?.(ref, event, run); return { event, run }; } diff --git a/src/index.ts b/src/index.ts index 06bb290..ffe1037 100644 --- a/src/index.ts +++ b/src/index.ts @@ -28,6 +28,13 @@ export * from "./engine/stageManager.js"; // Adapters export * from "./adapters/simulated.js"; export * from "./adapters/git.js"; +export * from "./adapters/process.js"; + +// The tracker layer (tickets, cross-task deps, state model) + HTTP API +export * from "./tracker/ticket.js"; +export * from "./tracker/store.js"; +export * from "./tracker/tracker.js"; +export * from "./server.js"; // Appendix A + §8 presets export * as presets from "./presets.js"; diff --git a/src/presets.ts b/src/presets.ts index f480268..a9bf497 100644 --- a/src/presets.ts +++ b/src/presets.ts @@ -230,6 +230,19 @@ export function panelReview(k = 2): FlowSpec { }; } +/** + * The state_map for the INT design flow (§1.1: "each board carries a + * state_map renaming these onto its columns") — run nodes onto the + * tracker's eight-value union. + */ +export const INT_DESIGN_STATE_MAP: Record = { + design: "design", + design_check: "design", + design_review: "design_review", + implement: "in_progress", + review: "in_review", +}; + /** * A flow with no beckett-flow block is compiled from the cast exactly as * today (effort → gate), so existing filings never break (§3.2): low/medium diff --git a/src/server.ts b/src/server.ts new file mode 100644 index 0000000..dbcbf5f --- /dev/null +++ b/src/server.ts @@ -0,0 +1,192 @@ +/** + * The tracker API — the "store + API to drive it all" the review asked for. + * A dependency-free node:http server over the Tracker: every mutation the + * TypeScript API offers is reachable over HTTP, so nothing requires + * importing the library to operate a board. + * + * GET /health + * GET /tickets → all tickets (the board) + * POST /tickets → file (FileTicketInput body) + * GET /tickets/:ref → TicketStatus (ticket + run + deps) + * POST /tickets/:ref/staff → open the run for a todo ticket + * POST /tickets/:ref/nudge → {text, node?} + * POST /tickets/:ref/pause + * POST /tickets/:ref/resume → {grant?} + * POST /tickets/:ref/gate → {node, verdict, note?} + * POST /tickets/:ref/cancel → {reason?} + * GET /tickets/:ref/events?tail=N → run event log + * GET /tickets/:ref/journal?tail=N → human narrative + * + * Refs arrive URL-encoded ("#3.1" → %233.1). Errors are {error} with + * 400/404/409. The server owns the engine tick cadence (≤30s, §6). + */ +import * as http from "node:http"; +import type { AddressInfo } from "node:net"; +import { Tracker } from "./tracker/tracker.js"; + +const TICKET_ROUTE = /^\/tickets\/([^/]+)(?:\/([a-z]+))?$/; + +export interface TrackerServerOptions { + /** engine tick cadence in ms; ≤30s keeps the §6 announce invariant */ + tickMs?: number; +} + +export class TrackerServer { + readonly server: http.Server; + private ticker: NodeJS.Timeout | null = null; + + constructor( + readonly tracker: Tracker, + private readonly opts: TrackerServerOptions = {}, + ) { + this.server = http.createServer((req, res) => { + void this.route(req, res).catch((err) => { + send(res, 500, { error: err instanceof Error ? err.message : String(err) }); + }); + }); + } + + async listen(port = 0, host = "127.0.0.1"): Promise { + await new Promise((resolve, reject) => { + this.server.once("error", reject); + this.server.listen(port, host, () => resolve()); + }); + const tickMs = this.opts.tickMs ?? 15_000; + this.ticker = setInterval(() => void this.tracker.tick(), tickMs); + this.ticker.unref(); + return (this.server.address() as AddressInfo).port; + } + + async close(): Promise { + if (this.ticker) clearInterval(this.ticker); + await new Promise((resolve, reject) => + this.server.close((err) => (err ? reject(err) : resolve())), + ); + } + + private async route(req: http.IncomingMessage, res: http.ServerResponse): Promise { + const url = new URL(req.url ?? "/", "http://localhost"); + const method = req.method ?? "GET"; + + if (method === "GET" && url.pathname === "/health") { + return send(res, 200, { ok: true }); + } + if (url.pathname === "/tickets") { + if (method === "GET") return send(res, 200, { tickets: this.tracker.list() }); + if (method === "POST") { + try { + const body = await readJson(req); + const ticket = await this.tracker.file(body as never); + return send(res, 201, { ticket }); + } catch (err) { + return send(res, 400, { error: message(err) }); + } + } + return send(res, 405, { error: "method not allowed" }); + } + + const match = TICKET_ROUTE.exec(url.pathname); + if (!match) return send(res, 404, { error: "not found" }); + const ref = decodeURIComponent(match[1]!); + const action = match[2]; + + try { + this.tracker.get(ref); + } catch { + return send(res, 404, { error: `no ticket ${ref}` }); + } + + try { + if (action == null) { + if (method !== "GET") return send(res, 405, { error: "method not allowed" }); + return send(res, 200, this.tracker.status(ref)); + } + if (method === "GET" && action === "events") { + const events = this.tracker.engine.store.readEvents(ref); + const tail = Number(url.searchParams.get("tail") ?? events.length); + return send(res, 200, { events: events.slice(-tail) }); + } + if (method === "GET" && action === "journal") { + const lines = this.tracker.engine.store.readJournal(ref); + const tail = Number(url.searchParams.get("tail") ?? lines.length); + return send(res, 200, { journal: lines.slice(-tail) }); + } + if (method !== "POST") return send(res, 405, { error: "method not allowed" }); + const body = (await readJson(req)) as Record; + switch (action) { + case "staff": + return send(res, 200, { ticket: await this.tracker.staff(ref) }); + case "nudge": { + if (typeof body["text"] !== "string") return send(res, 400, { error: "text required" }); + const receipt = this.tracker.nudge( + ref, + body["text"], + typeof body["node"] === "string" ? body["node"] : undefined, + ); + return send(res, 200, { receipt: receipt.receipt }); + } + case "pause": + return send(res, 200, { ticket: await this.tracker.pause(ref) }); + case "resume": + return send(res, 200, { + ticket: await this.tracker.resume(ref, body["grant"] as never), + }); + case "gate": { + const { node, verdict, note } = body as { + node?: string; + verdict?: string; + note?: string; + }; + if (typeof node !== "string" || (verdict !== "pass" && verdict !== "fail")) { + return send(res, 400, { error: "node and verdict (pass|fail) required" }); + } + return send(res, 200, { + ticket: await this.tracker.decideGate(ref, node, verdict, note), + }); + } + case "cancel": + return send(res, 200, { + ticket: await this.tracker.cancel( + ref, + typeof body["reason"] === "string" ? body["reason"] : undefined, + ), + }); + default: + return send(res, 404, { error: `unknown action ${action}` }); + } + } catch (err) { + const msg = message(err); + const status = /no ticket|does not exist/.test(msg) + ? 404 + : /already|is (done|cancelled|parked|running)|not parked|not running|no run yet|still waits/.test(msg) + ? 409 + : 400; + return send(res, status, { error: msg }); + } + } +} + +function message(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} + +function send(res: http.ServerResponse, status: number, body: unknown): void { + const payload = JSON.stringify(body); + res.writeHead(status, { + "content-type": "application/json", + "content-length": Buffer.byteLength(payload), + }); + res.end(payload); +} + +async function readJson(req: http.IncomingMessage): Promise { + const chunks: Buffer[] = []; + for await (const chunk of req) chunks.push(chunk as Buffer); + const raw = Buffer.concat(chunks).toString("utf8"); + if (!raw.trim()) return {}; + try { + return JSON.parse(raw); + } catch { + throw new Error("body is not valid JSON"); + } +} diff --git a/src/tracker/store.ts b/src/tracker/store.ts new file mode 100644 index 0000000..10d76f1 --- /dev/null +++ b/src/tracker/store.ts @@ -0,0 +1,120 @@ +/** + * The ticket registry — tasks.json: "a local, locked, versioned registry" + * (§1.5). Same durability posture as the run store: atomic tmp+rename + * writes, a version counter bumped on every write for optimistic + * concurrency, and an advisory directory lock so two processes never + * interleave a read-modify-write. + */ +import * as fs from "node:fs"; +import * as path from "node:path"; +import type { Ticket } from "./ticket.js"; + +export interface TasksFile { + /** bumped on every write; readers can detect concurrent mutation */ + version: number; + /** allocator for top-level "#N" refs */ + nextId: number; + /** allocator for "#N.x" child refs, keyed by parent ref */ + nextChild: Record; + tickets: Record; +} + +const EMPTY: TasksFile = { version: 0, nextId: 1, nextChild: {}, tickets: {} }; + +export class VersionConflictError extends Error { + constructor(expected: number, found: number) { + super(`tasks.json version conflict: expected ${expected}, found ${found}`); + this.name = "VersionConflictError"; + } +} + +export class TicketStore { + readonly file: string; + private readonly lockDir: string; + + constructor(root: string) { + fs.mkdirSync(root, { recursive: true }); + this.file = path.join(root, "tasks.json"); + this.lockDir = path.join(root, "tasks.json.lock"); + } + + read(): TasksFile { + if (!fs.existsSync(this.file)) return structuredClone(EMPTY); + const parsed = JSON.parse(fs.readFileSync(this.file, "utf8")) as TasksFile; + if (typeof parsed.version !== "number" || typeof parsed.tickets !== "object") { + throw new Error(`tasks.json is corrupt at ${this.file}`); + } + return parsed; + } + + /** + * Locked read-modify-write. `mutate` receives the current file and + * returns the value to expose to the caller; the (possibly mutated) file + * is written back atomically with the version bumped. + */ + update(mutate: (tasks: TasksFile) => T): T { + this.acquireLock(); + try { + const tasks = this.read(); + const before = tasks.version; + const result = mutate(tasks); + const onDisk = fs.existsSync(this.file) + ? (JSON.parse(fs.readFileSync(this.file, "utf8")) as TasksFile).version + : 0; + if (onDisk !== before) throw new VersionConflictError(before, onDisk); + tasks.version = before + 1; + const tmp = `${this.file}.tmp`; + fs.writeFileSync(tmp, JSON.stringify(tasks, null, 2)); + fs.renameSync(tmp, this.file); + return result; + } finally { + this.releaseLock(); + } + } + + get(ref: string): Ticket | undefined { + return this.read().tickets[ref]; + } + + list(): Ticket[] { + return Object.values(this.read().tickets); + } + + private acquireLock(): void { + const deadline = Date.now() + 5_000; + for (;;) { + try { + fs.mkdirSync(this.lockDir); + return; + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== "EEXIST") throw err; + // A crashed holder leaves a stale lock; break it after 10s of age. + try { + const age = Date.now() - fs.statSync(this.lockDir).mtimeMs; + if (age > 10_000) { + fs.rmdirSync(this.lockDir); + continue; + } + } catch { + continue; // raced with the holder's release — retry + } + if (Date.now() > deadline) { + throw new Error(`could not lock ${this.file} within 5s`); + } + // spin briefly; contention is rare and short (single-write registry) + const until = Date.now() + 25; + while (Date.now() < until) { + /* busy-wait a hair — keeps the store dependency-free and sync */ + } + } + } + } + + private releaseLock(): void { + try { + fs.rmdirSync(this.lockDir); + } catch { + // already released / broken by a peer — nothing to do + } + } +} diff --git a/src/tracker/ticket.ts b/src/tracker/ticket.ts new file mode 100644 index 0000000..61bddc5 --- /dev/null +++ b/src/tracker/ticket.ts @@ -0,0 +1,78 @@ +/** + * The ticket entity — the tracker layer bored was missing. A real ticket + * with the eight-value state union from §1.1 of the design doc (backlog, + * todo, design, design_review, in_progress, in_review, done, cancelled — + * done/cancelled terminal), the #N / #N.x tree with `needs` cross-task + * dependencies from §1.5, and a per-ticket state_map renaming run nodes + * onto tracker columns (§1.1's "each board carries a state_map"). + */ +import { z } from "zod"; +import type { FlowSpec, HarnessSpec, NodeId } from "../spec/types.js"; +import { flowSpecSchema, harnessSpecSchema } from "../spec/schema.js"; + +export const TICKET_STATES = [ + "backlog", + "todo", + "design", + "design_review", + "in_progress", + "in_review", + "done", + "cancelled", +] as const; + +export type TicketState = (typeof TICKET_STATES)[number]; + +export const TERMINAL_STATES: readonly TicketState[] = ["done", "cancelled"]; + +export function isTerminal(state: TicketState): boolean { + return TERMINAL_STATES.includes(state); +} + +export interface Ticket { + /** "#N" for top-level, "#N.x" for children — the user-facing tree ref (§1.5) */ + ref: string; + title: string; + body?: string; + criteria?: string[]; + state: TicketState; + /** why the ticket sits where it does (park reason, unmet needs, …) */ + stateReason?: string; + originChannel?: string; + /** tree structure: "#N" for a "#N.x" child */ + parent?: string; + children: string[]; + /** cross-task dependencies: refs that must be done before this staffs */ + needs: string[]; + /** the run of show; when absent, compiled from `cast` at staffing (§3.2) */ + flow?: FlowSpec; + cast?: HarnessSpec; + /** node → tracker column; falls back to node-kind defaults */ + stateMap?: Record; + /** staff automatically once needs are met */ + autoStaff: boolean; + /** a run has been opened for this ticket (the run ref IS the ticket ref) */ + staffed: boolean; + createdAt: string; + updatedAt: string; +} + +export const ticketStateSchema = z.enum(TICKET_STATES); + +/** What a filing provides; everything else is allocated by the tracker. */ +export const fileTicketSchema = z + .object({ + title: z.string().min(1), + body: z.string().optional(), + criteria: z.array(z.string()).optional(), + originChannel: z.string().optional(), + parent: z.string().optional(), + needs: z.array(z.string()).optional(), + flow: flowSpecSchema.optional(), + cast: harnessSpecSchema.optional(), + stateMap: z.record(z.string(), ticketStateSchema).optional(), + autoStaff: z.boolean().optional(), + }) + .strict(); + +export type FileTicketInput = z.infer; diff --git a/src/tracker/tracker.ts b/src/tracker/tracker.ts new file mode 100644 index 0000000..b89aac0 --- /dev/null +++ b/src/tracker/tracker.ts @@ -0,0 +1,376 @@ +/** + * The Tracker — the layer that makes this a Plane replacement rather than + * just a dispatcher: real tickets with the todo/in_progress/in_review/done + * state model, the #N / #N.x tree, cross-task `needs` dependencies with + * promote-on-done (§1.5's "cross-ticket DAG promotion: blockedBy → promote + * on done"), and ticket states projected live from the run record via the + * per-ticket state_map (§1.1). + * + * The tracker owns a StageManager: filing a ready ticket opens a run; every + * run event re-projects the ticket's column; a run finishing promotes the + * tickets that were waiting on it. + */ +import { compileFromCast } from "../presets.js"; +import type { FlowSpec, HarnessSpec } from "../spec/types.js"; +import { mustLint } from "../spec/lint.js"; +import type { FlowRun } from "../run/fold.js"; +import type { RunEvent } from "../run/events.js"; +import { + StageManager, + type ResumeGrant, + type StageManagerOptions, +} from "../engine/stageManager.js"; +import type { + AnnounceSink, + MergeProvider, + NudgeReceipt, + SpawnAdapter, +} from "../engine/ports.js"; +import { + fileTicketSchema, + isTerminal, + type FileTicketInput, + type Ticket, + type TicketState, +} from "./ticket.js"; +import { TicketStore } from "./store.js"; + +const DEFAULT_CAST: HarnessSpec = { harness: "claude", model: "claude-sonnet-5", effort: "high" }; + +export interface TicketStatus { + ticket: Ticket; + run?: FlowRun; + /** unmet needs (with their current states) — why a backlog ticket waits */ + blockedOn: Array<{ ref: string; state: TicketState }>; + /** open tickets whose needs include this one */ + blocking: string[]; +} + +export class Tracker { + readonly engine: StageManager; + readonly tickets: TicketStore; + /** serialised async work (auto-staffing promoted tickets) */ + private work: Promise = Promise.resolve(); + + constructor( + root: string, + spawner: SpawnAdapter, + merger: MergeProvider, + announceSink: AnnounceSink, + opts: StageManagerOptions, + ) { + this.tickets = new TicketStore(root); + this.engine = new StageManager(root, spawner, merger, announceSink, { + ...opts, + onEvent: (ref, event, run) => { + this.handleRunEvent(ref, event, run); + opts.onEvent?.(ref, event, run); + }, + }); + } + + /** Await all queued background work (promotion staffing). */ + async settle(): Promise { + let last: Promise; + do { + last = this.work; + await last; + } while (last !== this.work); // staffing may queue more staffing + } + + // ── filing (§1.5: the #N / #N.x tree with needs) ──────────────────────── + + async file(input: FileTicketInput): Promise { + const parsed = fileTicketSchema.parse(input); + if (parsed.flow) mustLint(parsed.flow); // fail at filing, not staffing + const now = new Date().toISOString(); + + const ticket = this.tickets.update((tasks): Ticket => { + // Allocate the ref inside the lock so refs never collide. + let ref: string; + if (parsed.parent != null) { + const parent = tasks.tickets[parsed.parent]; + if (!parent) throw new Error(`parent ${parsed.parent} does not exist`); + if (parent.parent != null) { + throw new Error(`parent ${parsed.parent} is itself a child; the tree is #N / #N.x`); + } + const child = tasks.nextChild[parsed.parent] ?? 1; + tasks.nextChild[parsed.parent] = child + 1; + ref = `${parsed.parent}.${child}`; + parent.children.push(ref); + } else { + ref = `#${tasks.nextId}`; + tasks.nextId += 1; + } + + // Needs may only reference existing tickets — the cross-task DAG is + // therefore acyclic by construction (a new node cannot be depended on + // yet, so no edge can ever point forward). + const needs = parsed.needs ?? []; + for (const need of needs) { + const target = tasks.tickets[need]; + if (!target) throw new Error(`needs ${need}: no such ticket`); + if (target.state === "cancelled") { + throw new Error(`needs ${need}: ticket is cancelled and will never complete`); + } + } + const unmet = needs.filter((n) => tasks.tickets[n]!.state !== "done"); + + const ticket: Ticket = { + ref, + title: parsed.title, + ...(parsed.body !== undefined ? { body: parsed.body } : {}), + ...(parsed.criteria !== undefined ? { criteria: parsed.criteria } : {}), + state: unmet.length > 0 ? "backlog" : "todo", + ...(unmet.length > 0 + ? { stateReason: `waiting on ${unmet.join(", ")}` } + : {}), + ...(parsed.originChannel !== undefined ? { originChannel: parsed.originChannel } : {}), + ...(parsed.parent !== undefined ? { parent: parsed.parent } : {}), + children: [], + needs, + ...(parsed.flow !== undefined ? { flow: parsed.flow as FlowSpec } : {}), + ...(parsed.cast !== undefined ? { cast: parsed.cast as HarnessSpec } : {}), + ...(parsed.stateMap !== undefined + ? { stateMap: parsed.stateMap as Record } + : {}), + autoStaff: parsed.autoStaff ?? true, + staffed: false, + createdAt: now, + updatedAt: now, + }; + tasks.tickets[ref] = ticket; + return ticket; + }); + + if (ticket.state === "todo" && ticket.autoStaff) { + await this.staff(ticket.ref); + return this.get(ticket.ref); + } + return ticket; + } + + // ── staffing: a ready ticket becomes a run ────────────────────────────── + + async staff(ref: string): Promise { + const ticket = this.get(ref); + if (ticket.staffed) throw new Error(`${ref} is already staffed`); + if (isTerminal(ticket.state)) throw new Error(`${ref} is ${ticket.state}`); + const unmet = this.unmetNeeds(ticket); + if (unmet.length > 0) { + throw new Error(`${ref} still waits on ${unmet.map((u) => u.ref).join(", ")}`); + } + // A flow with no beckett-flow block is compiled from the cast (§3.2). + const flow = ticket.flow ?? compileFromCast(ticket.cast ?? DEFAULT_CAST); + this.mutate(ref, (t) => { + t.staffed = true; + t.state = "in_progress"; + delete t.stateReason; + }); + await this.engine.open(ref, flow, { + ...(ticket.body !== undefined ? { body: ticket.body } : {}), + ...(ticket.criteria !== undefined ? { criteria: ticket.criteria } : {}), + ...(ticket.originChannel !== undefined ? { originChannel: ticket.originChannel } : {}), + }); + this.reproject(ref); + return this.get(ref); + } + + // ── reads ─────────────────────────────────────────────────────────────── + + get(ref: string): Ticket { + const ticket = this.tickets.get(ref); + if (!ticket) throw new Error(`no ticket ${ref}`); + return ticket; + } + + list(): Ticket[] { + return this.tickets.list(); + } + + status(ref: string): TicketStatus { + const ticket = this.get(ref); + const blocking = this.tickets + .list() + .filter((t) => t.needs.includes(ref) && !isTerminal(t.state)) + .map((t) => t.ref); + return { + ticket, + ...(ticket.staffed ? { run: this.engine.status(ref) } : {}), + blockedOn: this.unmetNeeds(ticket), + blocking, + }; + } + + // ── operator verbs (pass through to the run, then re-project) ────────── + + nudge(ref: string, text: string, node?: string): NudgeReceipt { + this.assertStaffed(ref); + return this.engine.nudge(ref, text, node); + } + + async pause(ref: string): Promise { + this.assertStaffed(ref); + await this.engine.pause(ref); + return this.get(ref); + } + + async resume(ref: string, grant?: ResumeGrant): Promise { + this.assertStaffed(ref); + await this.engine.resume(ref, grant); + return this.get(ref); + } + + async decideGate( + ref: string, + node: string, + verdict: "pass" | "fail", + note?: string, + ): Promise { + this.assertStaffed(ref); + await this.engine.decideHumanGate(ref, node, verdict, note); + return this.get(ref); + } + + async cancel(ref: string, reason?: string): Promise { + const ticket = this.get(ref); + if (ticket.staffed && !isTerminal(ticket.state)) { + await this.engine.cancel(ref, reason); + } else { + this.mutate(ref, (t) => { + t.state = "cancelled"; + if (reason !== undefined) t.stateReason = reason; + }); + } + return this.get(ref); + } + + async tick(): Promise { + await this.engine.tick(); + } + + /** Boot: replay runs, re-project every ticket, run missed promotions. */ + async recover(): Promise { + await this.engine.recoverAll(); + for (const ticket of this.tickets.list()) { + if (ticket.staffed) this.reproject(ticket.ref); + } + // Promotions that landed while the process was down. + for (const ticket of this.tickets.list()) { + if (ticket.state === "done") this.promoteDependents(ticket.ref); + } + await this.settle(); + } + + // ── projection: run record → ticket column ───────────────────────────── + + private handleRunEvent(ref: string, event: RunEvent, run: FlowRun): void { + if (!this.tickets.get(ref)) return; // runs opened outside the tracker + this.reprojectFrom(ref, run); + if (event.type === "run_done") this.promoteDependents(ref); + } + + private reproject(ref: string): void { + this.reprojectFrom(ref, this.engine.status(ref)); + } + + private reprojectFrom(ref: string, run: FlowRun): void { + const ticket = this.tickets.get(ref); + if (!ticket) return; + const projected = projectState(ticket, run); + if (ticket.state === projected.state && ticket.stateReason === projected.reason) return; + this.mutate(ref, (t) => { + t.state = projected.state; + if (projected.reason !== undefined) t.stateReason = projected.reason; + else delete t.stateReason; + }); + } + + /** + * §1.5: blockedBy → promote on done. Every open ticket waiting on `ref` + * whose needs are now all met moves backlog → todo; autoStaff tickets are + * staffed in the background (awaitable via settle()). + */ + private promoteDependents(doneRef: string): void { + const promoted: string[] = []; + for (const ticket of this.tickets.list()) { + if (ticket.staffed || isTerminal(ticket.state)) continue; + if (!ticket.needs.includes(doneRef)) continue; + const unmet = this.unmetNeeds(ticket); + if (unmet.length > 0) { + this.mutate(ticket.ref, (t) => { + t.stateReason = `waiting on ${unmet.map((u) => u.ref).join(", ")}`; + }); + continue; + } + this.mutate(ticket.ref, (t) => { + t.state = "todo"; + delete t.stateReason; + }); + promoted.push(ticket.ref); + } + for (const ref of promoted) { + const ticket = this.tickets.get(ref); + if (!ticket?.autoStaff) continue; + // Staffing opens a run (async); serialise it off the event path. + this.work = this.work.then(async () => { + const current = this.tickets.get(ref); + if (!current || current.staffed || isTerminal(current.state)) return; + await this.staff(ref); + }); + } + } + + private unmetNeeds(ticket: Ticket): Array<{ ref: string; state: TicketState }> { + return ticket.needs + .map((ref) => ({ ref, state: this.tickets.get(ref)?.state ?? ("cancelled" as TicketState) })) + .filter((n) => n.state !== "done"); + } + + private assertStaffed(ref: string): void { + if (!this.get(ref).staffed) throw new Error(`${ref} has no run yet (state: ${this.get(ref).state})`); + } + + private mutate(ref: string, fn: (t: Ticket) => void): void { + this.tickets.update((tasks) => { + const ticket = tasks.tickets[ref]; + if (!ticket) throw new Error(`no ticket ${ref}`); + fn(ticket); + ticket.updatedAt = new Date().toISOString(); + }); + } +} + +/** + * Project a run onto the tracker's eight-value state union. The ticket's + * state_map wins; otherwise node kind decides (workers/fanouts/joins are + * in_progress, gates are in_review), and any park a human must resolve + * reads as in_review — exactly today's "park in in_review for a human". + */ +export function projectState( + ticket: Ticket, + run: FlowRun, +): { state: TicketState; reason?: string } { + if (run.status === "done") return { state: "done" }; + if (run.status === "cancelled") { + return { state: "cancelled", ...(run.parked?.detail ? { reason: run.parked.detail } : {}) }; + } + const parkedReason = run.parked + ? `${run.parked.reason}${run.parked.detail ? `: ${run.parked.detail}` : ""}` + : undefined; + const activeNode = run.parked?.node ?? run.cursors[0]?.node; + if (activeNode != null) { + const mapped = ticket.stateMap?.[activeNode]; + if (mapped) return { state: mapped, ...(parkedReason ? { reason: parkedReason } : {}) }; + const def = run.spec.nodes[activeNode]; + if (def?.kind === "gate") { + return { state: "in_review", ...(parkedReason ? { reason: parkedReason } : {}) }; + } + if (run.status === "parked") { + return { state: "in_review", reason: parkedReason! }; + } + return { state: "in_progress" }; + } + if (run.status === "parked") return { state: "in_review", reason: parkedReason! }; + return { state: "in_progress" }; +} diff --git a/test/process-adapter.test.ts b/test/process-adapter.test.ts new file mode 100644 index 0000000..5139261 --- /dev/null +++ b/test/process-adapter.test.ts @@ -0,0 +1,254 @@ +/** + * The real agent-spawn adapter, end to end: the engine provisions a git + * worktree, spawns an actual node child process, the process reads the + * stage manifest, completes the §6.4 readiness handshake against real git + * state, does real work (writes + commits a file), emits its done-signal + * over the JSONL protocol — and the run completes. Plus the failure paths: + * silent exit, refusal, and nudges over stdin. + */ +import { execFileSync } from "node:child_process"; +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { describe, expect, it } from "vitest"; +import { + CollectingSink, + GitMergeProvider, + GitWorktreeSpawnAdapter, + ProcessSpawnAdapter, + StageManager, + Tracker, + coerceWorkerEvent, + presets, + systemClock, +} from "../src/index.js"; + +function git(cwd: string, ...args: string[]): string { + return execFileSync("git", args, { cwd, encoding: "utf8" }).trim(); +} + +function makeRepo(): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "ros-proc-repo-")); + git(dir, "init", "-q", "-b", "main"); + git(dir, "config", "user.email", "engine@test"); + git(dir, "config", "user.name", "engine"); + fs.writeFileSync(path.join(dir, "README.md"), "# repo\n"); + git(dir, "add", "-A"); + git(dir, "commit", "-q", "-m", "init"); + return dir; +} + +/** + * A minimal real worker: reads BECKETT_MANIFEST, proves where it is via + * git, writes a file, commits, and signals done. Behaviour switches on + * WORKER_MODE so the failure paths use the same binary. + */ +const WORKER_JS = ` +const fs = require("node:fs"); +const { execFileSync } = require("node:child_process"); +const emit = (o) => process.stdout.write(JSON.stringify(o) + "\\n"); +const git = (...a) => execFileSync("git", a, { encoding: "utf8" }).trim(); +const manifest = JSON.parse(fs.readFileSync(process.env.BECKETT_MANIFEST, "utf8")); +const mode = process.env.WORKER_MODE || "complete"; + +emit({ kind: "session_started" }); +if (mode === "refuse") { + emit({ kind: "worker_refused", observed: { branch: git("rev-parse", "--abbrev-ref", "HEAD"), reason: "env check failed: dirty state" } }); + process.exit(0); +} +emit({ + kind: "worker_ready", + manifestHash: manifest.manifestHash, + observedBranch: git("rev-parse", "--abbrev-ref", "HEAD"), + observedSha: git("rev-parse", "HEAD"), +}); +if (mode === "crash") process.exit(137); +if (mode === "hang-for-nudge") { + // wait for a nudge on stdin, echo it into the summary, then finish + let buf = ""; + process.stdin.on("data", (c) => { + buf += c.toString(); + const line = buf.split("\\n")[0]; + if (!line) return; + const msg = JSON.parse(line); + emit({ kind: "turn_completed", turn: 1, toolCalls: 1, tokens: { input: 10, output: 5 } }); + emit({ kind: "finished", signal: { status: "complete", summary: "heard: " + msg.text, filesChanged: [], checksRun: null, blockedReason: null }, spendUsd: 0.01 }); + process.exit(0); + }); + emit("waiting"); // non-protocol chatter — must be ignored + return; +} +// mode === "complete": do real work +emit({ kind: "turn_completed", turn: 1, toolCalls: 2, tokens: { input: 500, output: 200 } }); +fs.writeFileSync("output.txt", "made by a real worker for " + manifest.taskRef + "\\n"); +emit({ kind: "file_change", path: "output.txt" }); +git("add", "-A"); +git("commit", "-m", "worker output"); +emit({ kind: "checkpoint", sha: git("rev-parse", "HEAD") }); +emit({ + kind: "finished", + signal: { status: "complete", summary: "wrote output.txt", filesChanged: ["output.txt"], checksRun: null, blockedReason: null }, + spendUsd: 0.05, +}); +`; + +interface ProcRig { + engine: StageManager; + adapter: ProcessSpawnAdapter; + git: GitWorktreeSpawnAdapter; + repo: string; + sink: CollectingSink; + workerPath: string; +} + +function makeProcRig(env: Record = {}): ProcRig { + const repo = makeRepo(); + const workerPath = path.join(fs.mkdtempSync(path.join(os.tmpdir(), "ros-worker-")), "worker.cjs"); + fs.writeFileSync(workerPath, WORKER_JS); + const gitAdapter = new GitWorktreeSpawnAdapter(repo, systemClock); + const adapter = new ProcessSpawnAdapter(gitAdapter, () => ({ + cmd: process.execPath, + args: [workerPath], + env, + })); + const sink = new CollectingSink(); + const root = fs.mkdtempSync(path.join(os.tmpdir(), "ros-proc-store-")); + const engine = new StageManager(root, adapter, new GitMergeProvider(gitAdapter), sink, { + clock: systemClock, + ownerDM: "dm", + }); + adapter.connect((ref, seatKey, ev) => engine.deliverWorkerEvent(ref, seatKey, ev)); + return { engine, adapter, git: gitAdapter, repo, sink, workerPath }; +} + +async function waitFor(predicate: () => boolean, what: string, timeoutMs = 10_000): Promise { + const deadline = Date.now() + timeoutMs; + while (!predicate()) { + if (Date.now() > deadline) throw new Error(`timed out waiting for ${what}`); + await new Promise((r) => setTimeout(r, 25)); + } +} + +describe("ProcessSpawnAdapter — a real child process runs the work", () => { + it("spawns, handshakes against real git state, commits real output, completes the run", async () => { + const rig = makeProcRig(); + await rig.engine.open("#p1", presets.onePass(), { + body: "write output.txt", + criteria: ["output.txt exists"], + }); + await waitFor(() => rig.engine.status("#p1").status === "done", "run to complete"); + + const run = rig.engine.status("#p1"); + expect(run.spend.usd).toBeCloseTo(0.05); + const types = rig.engine.store.readEvents("#p1").map((e) => e.type); + expect(types).toContain("worker_ready"); // the handshake really verified + expect(types).toContain("checkpoint_committed"); + // the work is really on the task branch + const worktree = rig.git.taskWorktree("#p1"); + expect(fs.readFileSync(path.join(worktree, "output.txt"), "utf8")).toContain("#p1"); + expect(git(worktree, "log", "--format=%s", "-1")).toBe("worker output"); + // the manifest + brief were really written for the worker + expect(fs.existsSync(path.join(worktree, ".beckett", "stage-manifest.json"))).toBe(true); + const brief = fs.readFileSync(path.join(worktree, ".beckett", "brief.md"), "utf8"); + expect(brief).toContain("write output.txt"); + expect(brief).toContain("output.txt exists"); + }); + + it("a process dying without a done-signal synthesizes it, alarms, and re-staffs", async () => { + const rig = makeProcRig({ WORKER_MODE: "crash" }); + const spec = presets.onePass(); + spec.nodes["implement"]!.retries = 1; + await rig.engine.open("#p2", spec, { body: "b" }); + // attempt 1 crashes → retry → attempt 2 crashes → retries exhausted → park + await waitFor(() => rig.engine.status("#p2").status === "parked", "retries to exhaust"); + const run = rig.engine.status("#p2"); + expect(run.parked?.reason).toBe("retries_exhausted"); + const events = rig.engine.store.readEvents("#p2"); + expect(events.filter((e) => e.type === "seat_spawned")).toHaveLength(2); + expect( + events.some( + (e) => e.type === "alarm_raised" && (e as { alarm: { type: string } }).alarm.type === "silent_exit", + ), + ).toBe(true); + expect(rig.sink.pages().map((a) => a.eventType)).toContain("parked"); + }); + + it("a worker refusing its environment check walks the refusal path", async () => { + const rig = makeProcRig({ WORKER_MODE: "refuse" }); + const spec = presets.onePass(); + spec.nodes["implement"]!.retries = 0; + await rig.engine.open("#p3", spec, { body: "b" }); + await waitFor(() => rig.engine.status("#p3").status === "parked", "refusal to park"); + const events = rig.engine.store.readEvents("#p3").map((e) => e.type); + expect(events).toContain("worker_refused"); + }); + + it("nudges reach the live process over stdin; non-protocol stdout is ignored", async () => { + const rig = makeProcRig({ WORKER_MODE: "hang-for-nudge" }); + await rig.engine.open("#p4", presets.onePass(), { body: "b" }); + await waitFor( + () => rig.engine.store.readEvents("#p4").some((e) => e.type === "worker_ready"), + "worker to become ready", + ); + const receipt = rig.engine.nudge("#p4", "prefer approach B"); + expect(receipt.receipt).toBe("delivered"); + await waitFor(() => rig.engine.status("#p4").status === "done", "nudged worker to finish"); + const signal = rig.engine.store + .readEvents("#p4") + .find((e) => e.type === "signal_received"); + expect((signal as { signal: { summary: string } }).signal.summary).toBe( + "heard: prefer approach B", + ); + }); + + it("works end to end under the tracker (ticket → real process → done)", async () => { + const repo = makeRepo(); + const workerPath = path.join(fs.mkdtempSync(path.join(os.tmpdir(), "ros-worker-")), "worker.cjs"); + fs.writeFileSync(workerPath, WORKER_JS); + const gitAdapter = new GitWorktreeSpawnAdapter(repo, systemClock); + const adapter = new ProcessSpawnAdapter(gitAdapter, () => ({ + cmd: process.execPath, + args: [workerPath], + })); + const root = fs.mkdtempSync(path.join(os.tmpdir(), "ros-proc-tracker-")); + const tracker = new Tracker(root, adapter, new GitMergeProvider(gitAdapter), new CollectingSink(), { + clock: systemClock, + ownerDM: "dm", + }); + adapter.connect((ref, seatKey, ev) => tracker.engine.deliverWorkerEvent(ref, seatKey, ev)); + const ticket = await tracker.file({ title: "real work", body: "do it", flow: presets.onePass() }); + await waitFor(() => tracker.get(ticket.ref).state === "done", "ticket to reach done"); + const worktree = gitAdapter.taskWorktree(ticket.ref); + expect(fs.existsSync(path.join(worktree, "output.txt"))).toBe(true); + }); +}); + +describe("coerceWorkerEvent — the protocol boundary", () => { + it("accepts every well-formed kind and rejects malformed lines", () => { + expect(coerceWorkerEvent({ kind: "session_started" })).toEqual({ kind: "session_started" }); + expect( + coerceWorkerEvent({ kind: "worker_ready", manifestHash: "h", observedBranch: "b", observedSha: "s" }), + ).toMatchObject({ kind: "worker_ready" }); + expect(coerceWorkerEvent({ kind: "worker_ready", manifestHash: "h" })).toBeNull(); + expect(coerceWorkerEvent({ kind: "file_change", path: "x" })).toMatchObject({ path: "x" }); + expect(coerceWorkerEvent({ kind: "file_change" })).toBeNull(); + expect(coerceWorkerEvent({ kind: "unknown_thing" })).toBeNull(); + expect(coerceWorkerEvent("chatter")).toBeNull(); + expect(coerceWorkerEvent(null)).toBeNull(); + const finished = coerceWorkerEvent({ + kind: "finished", + signal: { status: "complete", summary: "s", filesChanged: ["a"], checksRun: null, blockedReason: null, data: { pass: true } }, + spendUsd: 1.5, + }); + expect(finished).toMatchObject({ + kind: "finished", + spendUsd: 1.5, + signal: { status: "complete", data: { pass: true } }, + }); + // a garbage signal degrades to null (→ the silent-exit path), never throws + expect(coerceWorkerEvent({ kind: "finished", signal: { status: "nonsense" } })).toMatchObject({ + kind: "finished", + signal: null, + }); + }); +}); diff --git a/test/server.test.ts b/test/server.test.ts new file mode 100644 index 0000000..ea2e778 --- /dev/null +++ b/test/server.test.ts @@ -0,0 +1,190 @@ +/** + * The HTTP API: every mutation is reachable over the wire — file, staff, + * nudge, gate, pause/resume, cancel — plus board and per-ticket reads. + * Runs against a live server on an ephemeral port with simulated workers. + */ +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { + CollectingSink, + ManualClock, + SimulatedMergeProvider, + SimulatedSpawnAdapter, + Tracker, + TrackerServer, + presets, +} from "../src/index.js"; + +interface ServerRig { + base: string; + server: TrackerServer; + tracker: Tracker; + adapter: SimulatedSpawnAdapter; +} + +let rig: ServerRig; + +async function call( + method: string, + pathname: string, + body?: unknown, +): Promise<{ status: number; data: Record }> { + const res = await fetch(`${rig.base}${pathname}`, { + method, + headers: { "content-type": "application/json" }, + ...(body !== undefined ? { body: JSON.stringify(body) } : {}), + }); + return { status: res.status, data: (await res.json()) as Record }; +} + +const enc = (ref: string) => encodeURIComponent(ref); + +beforeEach(async () => { + const clock = new ManualClock(); + const adapter = new SimulatedSpawnAdapter(clock); + const root = fs.mkdtempSync(path.join(os.tmpdir(), "ros-server-")); + const tracker = new Tracker(root, adapter, new SimulatedMergeProvider(), new CollectingSink(), { + clock, + ownerDM: "dm", + }); + adapter.connect((ref, seatKey, ev) => tracker.engine.deliverWorkerEvent(ref, seatKey, ev)); + const server = new TrackerServer(tracker, { tickMs: 60_000 }); + const port = await server.listen(0); + rig = { base: `http://127.0.0.1:${port}`, server, tracker, adapter }; +}); + +afterEach(async () => { + await rig.server.close(); +}); + +describe("the tracker API", () => { + it("health, file, board, ticket status", async () => { + expect((await call("GET", "/health")).data).toEqual({ ok: true }); + + const created = await call("POST", "/tickets", { + title: "api-filed work", + body: "over the wire", + criteria: ["works"], + flow: presets.onePass(), + }); + expect(created.status).toBe(201); + const ticket = created.data["ticket"] as { ref: string; state: string }; + expect(ticket.ref).toBe("#1"); + expect(ticket.state).toBe("in_progress"); // auto-staffed + + const board = await call("GET", "/tickets"); + expect((board.data["tickets"] as unknown[]).length).toBe(1); + + const status = await call("GET", `/tickets/${enc("#1")}`); + expect(status.status).toBe(200); + expect((status.data["run"] as { status: string }).status).toBe("running"); + expect(status.data["blockedOn"]).toEqual([]); + }); + + it("drives a full reviewed lifecycle over the wire, gate verdict included", async () => { + await call("POST", "/tickets", { + title: "human-gated", + body: "b", + flow: { + version: 1, + entry: "implement", + nodes: { + implement: { + kind: "worker", + cast: { harness: "pi", effort: "high" }, + onPass: "approve", + onFail: "park", + }, + approve: { kind: "gate", by: "human", onPass: "done", onFail: "implement" }, + }, + }, + }); + const seat = rig.adapter.seat("implement"); + await seat.ready(); + + // steer it over the wire + const nudged = await call("POST", `/tickets/${enc("#1")}/nudge`, { text: "small diff please" }); + expect(nudged.data["receipt"]).toBe("delivered"); + expect(seat.nudges).toContain("small diff please"); + + await seat.complete(); + // parked at the human gate → in_review column + const parked = await call("GET", `/tickets/${enc("#1")}`); + expect((parked.data["ticket"] as { state: string }).state).toBe("in_review"); + + const gated = await call("POST", `/tickets/${enc("#1")}/gate`, { + node: "approve", + verdict: "pass", + note: "lgtm", + }); + expect(gated.status).toBe(200); + expect((gated.data["ticket"] as { state: string }).state).toBe("done"); + + const events = await call("GET", `/tickets/${enc("#1")}/events?tail=3`); + expect((events.data["events"] as Array<{ type: string }>).map((e) => e.type)).toContain( + "run_done", + ); + const journal = await call("GET", `/tickets/${enc("#1")}/journal?tail=1`); + expect((journal.data["journal"] as string[])[0]).toContain("run done"); + }); + + it("pause and resume over the wire", async () => { + await call("POST", "/tickets", { title: "pausable", body: "b", flow: presets.onePass() }); + const seat = rig.adapter.seat("implement"); + await seat.ready(); + const paused = await call("POST", `/tickets/${enc("#1")}/pause`); + expect((paused.data["ticket"] as { state: string }).state).toBe("in_review"); + const resumed = await call("POST", `/tickets/${enc("#1")}/resume`, {}); + expect((resumed.data["ticket"] as { state: string }).state).toBe("in_progress"); + const seat2 = rig.adapter.seat("implement", { attempt: 2 }); + await seat2.ready(); + await seat2.complete(); + expect(rig.tracker.get("#1").state).toBe("done"); + }); + + it("cancel over the wire", async () => { + await call("POST", "/tickets", { title: "doomed", body: "b", flow: presets.onePass() }); + const cancelled = await call("POST", `/tickets/${enc("#1")}/cancel`, { reason: "descoped" }); + expect((cancelled.data["ticket"] as { state: string }).state).toBe("cancelled"); + }); + + it("filing with needs over the wire; dependents promote when the API completes work", async () => { + await call("POST", "/tickets", { title: "a", body: "b", flow: presets.onePass() }); + const dep = await call("POST", "/tickets", { + title: "b", + body: "b", + flow: presets.onePass(), + needs: ["#1"], + }); + expect((dep.data["ticket"] as { state: string }).state).toBe("backlog"); + const seat = rig.adapter.seat("implement"); + await seat.ready(); + await seat.complete(); + await rig.tracker.settle(); + const promoted = await call("GET", `/tickets/${enc("#2")}`); + expect((promoted.data["ticket"] as { state: string }).state).toBe("in_progress"); + }); + + it("maps errors to 400/404/409", async () => { + expect((await call("GET", `/tickets/${enc("#42")}`)).status).toBe(404); + expect((await call("POST", "/tickets", { title: "" })).status).toBe(400); + expect((await call("POST", "/tickets", { titel: "typo" })).status).toBe(400); + + await call("POST", "/tickets", { title: "x", body: "b", flow: presets.onePass() }); + // staffing an already-staffed ticket conflicts + expect((await call("POST", `/tickets/${enc("#1")}/staff`)).status).toBe(409); + // resuming a running ticket conflicts + expect((await call("POST", `/tickets/${enc("#1")}/resume`, {})).status).toBe(409); + // gate verdict validation + expect( + (await call("POST", `/tickets/${enc("#1")}/gate`, { node: "x", verdict: "maybe" })).status, + ).toBe(400); + // unknown action + expect((await call("POST", `/tickets/${enc("#1")}/frobnicate`, {})).status).toBe(404); + // non-JSON body + const res = await fetch(`${rig.base}/tickets`, { method: "POST", body: "not json{" }); + expect(res.status).toBe(400); + }); +}); diff --git a/test/tracker.test.ts b/test/tracker.test.ts new file mode 100644 index 0000000..f4c607a --- /dev/null +++ b/test/tracker.test.ts @@ -0,0 +1,283 @@ +/** + * The tracker layer: the ticket entity, the #N / #N.x tree, cross-task + * `needs` with promote-on-done, the eight-value state model projected from + * runs, and the locked/versioned tasks.json registry. + */ +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { describe, expect, it } from "vitest"; +import { + CollectingSink, + ManualClock, + SimulatedMergeProvider, + SimulatedSpawnAdapter, + TicketStore, + Tracker, + presets, +} from "../src/index.js"; + +interface TrackerRig { + tracker: Tracker; + adapter: SimulatedSpawnAdapter; + sink: CollectingSink; + clock: ManualClock; + root: string; +} + +function makeTrackerRig(): TrackerRig { + const clock = new ManualClock(); + const adapter = new SimulatedSpawnAdapter(clock); + const sink = new CollectingSink(); + const root = fs.mkdtempSync(path.join(os.tmpdir(), "ros-tracker-")); + const tracker = new Tracker(root, adapter, new SimulatedMergeProvider(), sink, { + clock, + ownerDM: "dm:owner", + }); + adapter.connect((ref, seatKey, ev) => tracker.engine.deliverWorkerEvent(ref, seatKey, ev)); + return { tracker, adapter, sink, clock, root }; +} + +async function completeOnePass(rig: TrackerRig, ref: string): Promise { + const seat = [...rig.adapter.seats] + .reverse() + .find((s) => s.request.ref === ref && s.request.node === "implement"); + expect(seat, `a seat for ${ref}`).toBeTruthy(); + await seat!.ready(); + await seat!.complete(); +} + +const onePassInput = (title: string, extra: Record = {}) => ({ + title, + body: `${title} body`, + flow: presets.onePass(), + ...extra, +}); + +describe("ticket store — tasks.json, locked and versioned", () => { + it("bumps the version on every write and survives re-reads", () => { + const store = new TicketStore(fs.mkdtempSync(path.join(os.tmpdir(), "ros-tstore-"))); + expect(store.read().version).toBe(0); + store.update((t) => { + t.nextId = 5; + }); + store.update((t) => { + t.nextId = 9; + }); + const tasks = store.read(); + expect(tasks.version).toBe(2); + expect(tasks.nextId).toBe(9); + }); +}); + +describe("filing and the #N / #N.x tree", () => { + it("allocates sequential top-level refs and child refs under a parent", async () => { + const rig = makeTrackerRig(); + const a = await rig.tracker.file(onePassInput("first", { autoStaff: false })); + const b = await rig.tracker.file(onePassInput("second", { autoStaff: false })); + expect(a.ref).toBe("#1"); + expect(b.ref).toBe("#2"); + const child1 = await rig.tracker.file(onePassInput("child", { parent: "#1", autoStaff: false })); + const child2 = await rig.tracker.file(onePassInput("child2", { parent: "#1", autoStaff: false })); + expect(child1.ref).toBe("#1.1"); + expect(child2.ref).toBe("#1.2"); + expect(rig.tracker.get("#1").children).toEqual(["#1.1", "#1.2"]); + expect(child1.parent).toBe("#1"); + // the tree is one level deep + await expect( + rig.tracker.file(onePassInput("grandchild", { parent: "#1.1" })), + ).rejects.toThrow(/itself a child/); + }); + + it("refuses needs on unknown or cancelled tickets", async () => { + const rig = makeTrackerRig(); + await expect(rig.tracker.file(onePassInput("x", { needs: ["#99"] }))).rejects.toThrow( + /no such ticket/, + ); + const a = await rig.tracker.file(onePassInput("a", { autoStaff: false })); + await rig.tracker.cancel(a.ref, "superseded"); + await expect(rig.tracker.file(onePassInput("b", { needs: [a.ref] }))).rejects.toThrow( + /cancelled/, + ); + }); + + it("refuses an invalid flow at filing time (the linter runs before anything is stored)", async () => { + const rig = makeTrackerRig(); + await expect( + rig.tracker.file({ + title: "bad", + flow: { version: 1, entry: "ghost", nodes: {} }, + }), + ).rejects.toThrow(/bad_entry/); + expect(rig.tracker.list()).toHaveLength(0); + }); +}); + +describe("cross-task dependencies — blockedBy → promote on done (§1.5)", () => { + it("a ticket with unmet needs sits in backlog and staffs when its needs complete", async () => { + const rig = makeTrackerRig(); + const a = await rig.tracker.file(onePassInput("build the API")); + expect(rig.tracker.get(a.ref).state).toBe("in_progress"); // auto-staffed + const b = await rig.tracker.file(onePassInput("build the UI", { needs: [a.ref] })); + expect(b.state).toBe("backlog"); + expect(b.stateReason).toContain(a.ref); + expect(rig.adapter.seatCount()).toBe(1); // b burned zero tokens + + // status shows both directions of the dependency + const status = rig.tracker.status(b.ref); + expect(status.blockedOn).toEqual([{ ref: a.ref, state: "in_progress" }]); + expect(rig.tracker.status(a.ref).blocking).toEqual([b.ref]); + + await completeOnePass(rig, a.ref); + await rig.tracker.settle(); // promotion staffs b in the background + const promoted = rig.tracker.get(b.ref); + expect(promoted.state).toBe("in_progress"); + expect(promoted.staffed).toBe(true); + const bSeat = rig.adapter.seats[rig.adapter.seats.length - 1]!; + expect(bSeat.request.ref).toBe(b.ref); + await bSeat.ready(); + await bSeat.complete(); + expect(rig.tracker.get(b.ref).state).toBe("done"); + }); + + it("promotion waits for ALL needs; partial completion only updates the reason", async () => { + const rig = makeTrackerRig(); + const a = await rig.tracker.file(onePassInput("a")); + const b = await rig.tracker.file(onePassInput("b")); + const c = await rig.tracker.file(onePassInput("c", { needs: [a.ref, b.ref] })); + expect(c.state).toBe("backlog"); + await completeOnePass(rig, a.ref); + await rig.tracker.settle(); + const mid = rig.tracker.get(c.ref); + expect(mid.state).toBe("backlog"); + expect(mid.stateReason).toBe(`waiting on ${b.ref}`); + await completeOnePass(rig, b.ref); + await rig.tracker.settle(); + expect(rig.tracker.get(c.ref).state).toBe("in_progress"); + }); + + it("autoStaff:false tickets promote to todo and wait for a human staff call", async () => { + const rig = makeTrackerRig(); + const a = await rig.tracker.file(onePassInput("a")); + const b = await rig.tracker.file(onePassInput("b", { needs: [a.ref], autoStaff: false })); + await completeOnePass(rig, a.ref); + await rig.tracker.settle(); + expect(rig.tracker.get(b.ref).state).toBe("todo"); + expect(rig.tracker.get(b.ref).staffed).toBe(false); + await rig.tracker.staff(b.ref); + expect(rig.tracker.get(b.ref).state).toBe("in_progress"); + }); + + it("filing with met needs goes straight to todo/staffed", async () => { + const rig = makeTrackerRig(); + const a = await rig.tracker.file(onePassInput("a")); + await completeOnePass(rig, a.ref); + const b = await rig.tracker.file(onePassInput("b", { needs: [a.ref] })); + expect(b.staffed).toBe(true); + expect(b.state).toBe("in_progress"); + }); +}); + +describe("state projection — runs onto the eight-value union (§1.1)", () => { + it("walks todo → in_progress → in_review → done through the reviewed lifecycle", async () => { + const rig = makeTrackerRig(); + const t = await rig.tracker.file({ + title: "reviewed work", + body: "b", + flow: presets.reviewedLifecycle(), + }); + expect(rig.tracker.get(t.ref).state).toBe("in_progress"); + const seat = rig.adapter.seat("implement"); + await seat.ready(); + await seat.complete(); + expect(rig.tracker.get(t.ref).state).toBe("in_review"); // gate active + const review = rig.adapter.seat("review"); + await review.ready(); + await review.complete({ data: { pass: true } }); + expect(rig.tracker.get(t.ref).state).toBe("done"); + }); + + it("the INT design flow projects design / design_review via the state_map", async () => { + const rig = makeTrackerRig(); + const t = await rig.tracker.file({ + title: "INT work", + body: "b", + flow: presets.intDesignFlow(), + stateMap: presets.INT_DESIGN_STATE_MAP, + }); + expect(rig.tracker.get(t.ref).state).toBe("design"); + const design = rig.adapter.seat("design"); + await design.ready(); + await design.complete(); + expect(rig.tracker.get(t.ref).state).toBe("design"); // design_check maps to design + const check = rig.adapter.seat("design_check"); + await check.ready(); + await check.complete({ data: { pass: true } }); + // parked at the human design_review gate — the design_review column + expect(rig.tracker.get(t.ref).state).toBe("design_review"); + await rig.tracker.decideGate(t.ref, "design_review", "pass"); + expect(rig.tracker.get(t.ref).state).toBe("in_progress"); + }); + + it("parks a human owns read as in_review, with the reason on the ticket", async () => { + const rig = makeTrackerRig(); + const t = await rig.tracker.file(onePassInput("blocked work")); + const seat = rig.adapter.seat("implement"); + await seat.ready(); + await seat.blocked("needs credentials"); + const after = rig.tracker.get(t.ref); + expect(after.state).toBe("in_review"); + expect(after.stateReason).toContain("needs credentials"); + }); + + it("cancel mid-run cancels the run and the ticket", async () => { + const rig = makeTrackerRig(); + const t = await rig.tracker.file(onePassInput("doomed")); + const seat = rig.adapter.seat("implement"); + await seat.ready(); + await rig.tracker.cancel(t.ref, "descoped"); + expect(rig.tracker.get(t.ref).state).toBe("cancelled"); + expect(rig.tracker.engine.status(t.ref).status).toBe("cancelled"); + }); + + it("a ticket with no flow compiles one from the cast (§3.2), defaulting sanely", async () => { + const rig = makeTrackerRig(); + const low = await rig.tracker.file({ title: "trivial", cast: { harness: "pi", effort: "low" } }); + // low effort → self tier → one pass, no reviewer + const run = rig.tracker.engine.status(low.ref); + expect(Object.keys(run.spec.nodes)).toEqual(["implement"]); + const noCast = await rig.tracker.file({ title: "default cast" }); + const run2 = rig.tracker.engine.status(noCast.ref); + expect(Object.keys(run2.spec.nodes)).toEqual(["implement", "review"]); // fresh review tier + }); +}); + +describe("tracker recovery", () => { + it("re-projects states and runs missed promotions on boot", async () => { + const rig = makeTrackerRig(); + const a = await rig.tracker.file(onePassInput("a")); + const b = await rig.tracker.file(onePassInput("b", { needs: [a.ref], autoStaff: false })); + await completeOnePass(rig, a.ref); + await rig.tracker.settle(); + expect(rig.tracker.get(b.ref).state).toBe("todo"); + // simulate the promotion write being lost in a crash: rewind to backlog + rig.tracker.tickets.update((tasks) => { + tasks.tickets[b.ref]!.state = "backlog"; + tasks.tickets[b.ref]!.stateReason = `waiting on ${a.ref}`; + }); + + // reboot over the same root + const clock = new ManualClock(); + const adapter = new SimulatedSpawnAdapter(clock); + const rebooted = new Tracker(rig.root, adapter, new SimulatedMergeProvider(), new CollectingSink(), { + clock, + ownerDM: "dm:owner", + }); + adapter.connect((ref, seatKey, ev) => rebooted.engine.deliverWorkerEvent(ref, seatKey, ev)); + await rebooted.recover(); + const recovered = rebooted.get(b.ref); + expect(recovered.state).toBe("todo"); // promotion re-derived from a's done state + expect(recovered.staffed).toBe(false); // autoStaff:false respected on replay + expect(rebooted.get(a.ref).state).toBe("done"); + }); +}); From 002da8f434d4c37ed645a9a5814ec66c4cc8b1a3 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 13 Jul 2026 14:25:40 +0000 Subject: [PATCH 2/2] Add JS workflow authoring: fluent builder, flow scripts, and hooks Workflows can now be authored with JavaScript instead of (or alongside) JSON, so the shape is computed for the task at hand rather than one rigid structure for every ticket: - FlowBuilder: fluent flow().worker().gate().fanout().join().budget() .build() API; build() runs the full linter so an unbounded or malformed graph never leaves the builder - Flow scripts: a .js/.mjs/.cjs file default-exports (ctx) => flow, where ctx carries the ticket (title/body/criteria), the builder, and the presets; supports async scripts, wrapped returns ({flow, stateMap, hooks}), module-level exports, and CJS interop; results are linted and frozen exactly like hand-written specs - Hooks: an onEvent export acts as a scripted concierge with operator authority (nudge, pause/resume with grants, gate verdicts, cancel, file follow-up tickets); runs serialised off the event path; errors are journalled, never fatal; reloaded from the stored script path on recovery - Wiring: Ticket.flowScript + filing via the tracker, HTTP API and CLI (--flow-script); examples/flows/adaptive.flow.mjs demonstrates per-task shapes (hotfix -> one-pass, "areas:" -> computed fanout arms, criteria count -> effort) plus an auto-extend-once park policy - 15 new tests (166 total): builder composition/refusals, script loading (async, cjs, error paths), per-task shape divergence through the tracker, hook policies (auto-approve gate, auto-grant on park, follow-up filing, error journalling), and hook reload across recovery Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01A3DUAdiNuM62Q6TVAyj7rv --- README.md | 45 ++++ examples/flows/adaptive.flow.mjs | 57 ++++++ src/authoring/builder.ts | 152 ++++++++++++++ src/authoring/script.ts | 156 ++++++++++++++ src/cli.ts | 6 +- src/index.ts | 4 + src/tracker/ticket.ts | 4 + src/tracker/tracker.ts | 89 +++++++- test/authoring.test.ts | 341 +++++++++++++++++++++++++++++++ 9 files changed, 852 insertions(+), 2 deletions(-) create mode 100644 examples/flows/adaptive.flow.mjs create mode 100644 src/authoring/builder.ts create mode 100644 src/authoring/script.ts create mode 100644 test/authoring.test.ts diff --git a/README.md b/README.md index b80dbc1..20ef73e 100644 --- a/README.md +++ b/README.md @@ -96,6 +96,51 @@ the branch/sha it observes via git), emits progress events on stdout, and finishes with a done-signal. A process that dies silently is alarmed, retried, and parked by the engine — never lost. +## Authoring workflows in JavaScript + +JSON specs and presets still work, but flows can be *scripted*: a `.js`/ +`.mjs`/`.cjs` file default-exports a function that receives the ticket and +returns the flow — so the shape is computed for the task at hand instead of +one rigid structure for everything: + +```js +// adaptive.flow.mjs (full version in examples/flows/) +export default ({ ticket, flow, presets }) => { + if (/hotfix/i.test(ticket.title)) return presets.onePass(); + const areas = /areas:\s*(.+)/i.exec(ticket.body ?? "")?.[1]?.split(","); + if (areas) { + return flow() + .fanout("split", { + arms: areas.map((a) => ({ cast: { harness: "pi", effort: "high" }, brief: `own ${a.trim()}` })), + join: "land", + }) + .join("land", { strategy: "all-merge", onPass: "review" }) + .gate("review", { by: { cast: { harness: "claude", model: "claude-sonnet-5" }, rubric: "criteria-vs-diff" }, onPass: "done" }) + .budget({ usd: 40 }) + .build(); + } + return presets.reviewedLifecycle(); +}; + +// hooks: a scripted concierge with operator authority, per task +export const hooks = { + async onEvent({ event, actions }) { + if (event.type === "parked" && event.reason === "max_visits_exhausted") { + await actions.resume({ extraVisits: 1 }); // one free extension, then a human owns it + } + }, +}; +``` + +File it with `flowScript` (API), `--flow-script` (CLI), or +`tracker.file({..., flowScript})`. Scripting controls shape and management — +never the execution guarantees: the returned flow goes through the full +linter (bounded loops, closed node algebra), hooks run off the event path +with the same verbs a human concierge has, and budgets/caps still fence +everything, so scripted flows should set a budget. Hooks are reloaded from +the stored script path on recovery; hook errors are journalled, never fatal. +The fluent `flow()` builder is also exported for TypeScript callers. + ## Using it as a primitive ```ts diff --git a/examples/flows/adaptive.flow.mjs b/examples/flows/adaptive.flow.mjs new file mode 100644 index 0000000..fb2e46c --- /dev/null +++ b/examples/flows/adaptive.flow.mjs @@ -0,0 +1,57 @@ +/** + * An adaptive flow script: the workflow's shape is computed for the ticket + * at hand instead of one rigid structure for every task. + * + * - "hotfix" in the title → one pass, low effort, tight budget + * - "areas:" in the body → one fanout arm per listed area, all-merge, + * then a fresh review of the joined diff + * - three or more criteria → the full reviewed lifecycle, high effort + * - otherwise → reviewed lifecycle, medium effort + * + * The hooks below act as a scripted concierge: one automatic extra visit + * when a rework loop parks, and an auto-filed follow-up ticket when a run + * finishes with leftover TODOs mentioned in its park detail. + */ +export default ({ ticket, flow, presets }) => { + const title = ticket.title ?? ""; + const body = ticket.body ?? ""; + + if (/hotfix/i.test(title)) { + const spec = presets.onePass({ harness: "pi", effort: "low" }); + spec.budget = { usd: 2 }; + return spec; + } + + const areas = /areas:\s*([^\n]+)/i.exec(body)?.[1]; + if (areas) { + const arms = areas.split(",").map((area) => ({ + cast: { harness: "pi", effort: "high" }, + brief: `own the ${area.trim()} half of this task`, + })); + return flow() + .fanout("split", { arms, join: "land" }) + .join("land", { strategy: "all-merge", onPass: "review" }) + .gate("review", { + by: { + cast: { harness: "claude", model: "claude-sonnet-5", effort: "high" }, + rubric: "criteria-vs-diff, every area present", + }, + onPass: "done", + onFail: "park", + }) + .budget({ usd: 40, maxConcurrent: Math.min(arms.length, 4) }) + .build(); + } + + const effort = (ticket.criteria?.length ?? 0) >= 3 ? "high" : "medium"; + return presets.reviewedLifecycle({ harness: "pi", effort }); +}; + +export const hooks = { + async onEvent({ event, actions }) { + // one free extension when the rework loop exhausts — then a human owns it + if (event.type === "parked" && event.reason === "max_visits_exhausted") { + await actions.resume({ extraVisits: 1 }); + } + }, +}; diff --git a/src/authoring/builder.ts b/src/authoring/builder.ts new file mode 100644 index 0000000..5dad5b8 --- /dev/null +++ b/src/authoring/builder.ts @@ -0,0 +1,152 @@ +/** + * The flow builder — scripted authoring for the run of show. Instead of + * hand-writing JSON, a script (or any TypeScript caller) composes the DAG + * fluently and gets a linted FlowSpec back. The output is exactly the same + * closed algebra the engine executes — scripting controls the *shape* per + * task; execution semantics stay bounded and deterministic (§3.1's fence + * against the workflow-engine tarpit is preserved: build() runs the full + * linter, so an unbounded or malformed graph never leaves the builder). + */ +import { mustLint, type LintOptions } from "../spec/lint.js"; +import type { + FanoutArm, + FlowBudget, + FlowNode, + FlowSpec, + HarnessSpec, + JoinStrategy, + NodeId, + SuperviseSpec, +} from "../spec/types.js"; + +export interface WorkerOpts { + cast: HarnessSpec; + brief?: string; + artifact?: string; + onPass: NodeId | "done"; + /** defaults to "park" — hand to a human */ + onFail?: NodeId | "park"; + maxVisits?: number; + retries?: number; +} + +export interface GateOpts { + by: "human" | { cast: HarnessSpec; rubric: string }; + onPass: NodeId | "done"; + onFail?: NodeId | "park"; + maxFails?: number; + maxVisits?: number; +} + +export interface FanoutOpts { + arms: FanoutArm[]; + /** defaults to "worktree-each" — writers isolate */ + isolation?: "worktree-each" | "shared"; + join: NodeId; + maxVisits?: number; + retries?: number; +} + +export interface JoinOpts { + strategy: JoinStrategy; + quorumK?: number; + onPass: NodeId | "done"; + onFail?: NodeId | "park"; + maxVisits?: number; +} + +export class FlowBuilder { + private readonly nodes: Record = {}; + private entryId: NodeId | undefined; + private budgetSpec: FlowBudget | undefined; + private superviseSpec: SuperviseSpec | undefined; + + /** Explicit entry; otherwise the first node added is the entry. */ + entry(id: NodeId): this { + this.entryId = id; + return this; + } + + private add(id: NodeId, node: FlowNode): this { + if (this.nodes[id]) throw new Error(`node "${id}" is already defined`); + this.nodes[id] = node; + this.entryId ??= id; + return this; + } + + worker(id: NodeId, opts: WorkerOpts): this { + return this.add(id, { + kind: "worker", + cast: opts.cast, + ...(opts.brief !== undefined ? { brief: opts.brief } : {}), + ...(opts.artifact !== undefined ? { artifact: opts.artifact } : {}), + onPass: opts.onPass, + onFail: opts.onFail ?? "park", + ...(opts.maxVisits !== undefined ? { maxVisits: opts.maxVisits } : {}), + ...(opts.retries !== undefined ? { retries: opts.retries } : {}), + }); + } + + gate(id: NodeId, opts: GateOpts): this { + return this.add(id, { + kind: "gate", + by: opts.by, + onPass: opts.onPass, + onFail: opts.onFail ?? "park", + ...(opts.maxFails !== undefined ? { maxFails: opts.maxFails } : {}), + ...(opts.maxVisits !== undefined ? { maxVisits: opts.maxVisits } : {}), + }); + } + + fanout(id: NodeId, opts: FanoutOpts): this { + return this.add(id, { + kind: "fanout", + arms: opts.arms, + isolation: opts.isolation ?? "worktree-each", + join: opts.join, + ...(opts.maxVisits !== undefined ? { maxVisits: opts.maxVisits } : {}), + ...(opts.retries !== undefined ? { retries: opts.retries } : {}), + }); + } + + join(id: NodeId, opts: JoinOpts): this { + return this.add(id, { + kind: "join", + strategy: opts.strategy, + ...(opts.quorumK !== undefined ? { quorumK: opts.quorumK } : {}), + onPass: opts.onPass, + onFail: opts.onFail ?? "park", + ...(opts.maxVisits !== undefined ? { maxVisits: opts.maxVisits } : {}), + }); + } + + budget(budget: FlowBudget): this { + this.budgetSpec = budget; + return this; + } + + supervise(supervise: SuperviseSpec): this { + this.superviseSpec = supervise; + return this; + } + + /** Lint and freeze. Throws with every lint issue listed if the graph is bad. */ + build(lintOpts?: LintOptions): FlowSpec { + if (!this.entryId) throw new Error("an empty flow has no entry — add a node"); + const spec: FlowSpec = { + version: 1, + entry: this.entryId, + nodes: this.nodes, + ...(this.budgetSpec !== undefined ? { budget: this.budgetSpec } : {}), + ...(this.superviseSpec !== undefined ? { supervise: this.superviseSpec } : {}), + }; + return mustLint(spec, lintOpts); + } +} + +/** Entry point for scripts: `flow().worker("implement", {...}).build()`. */ +export function flow(entry?: NodeId): FlowBuilder { + const builder = new FlowBuilder(); + if (entry != null) builder.entry(entry); + return builder; +} diff --git a/src/authoring/script.ts b/src/authoring/script.ts new file mode 100644 index 0000000..c333421 --- /dev/null +++ b/src/authoring/script.ts @@ -0,0 +1,156 @@ +/** + * Flow scripts — workflows authored as JavaScript, not just JSON. A `.js`/ + * `.mjs`/`.cjs` file default-exports a (possibly async) function that + * receives the ticket context plus the authoring toolkit and returns the + * flow — so the shape is *computed for the task at hand* (branch on the + * title, derive fanout arms from data, compose presets) instead of one + * rigid structure for every ticket. + * + * // adaptive.flow.mjs + * export default ({ ticket, flow, presets }) => { + * if (/hotfix/i.test(ticket.title)) return presets.onePass(); + * return flow() + * .worker("implement", { cast: {...}, onPass: "review" }) + * .gate("review", { by: {...}, onPass: "done", onFail: "implement" }) + * .build(); + * }; + * export const stateMap = { implement: "in_progress", review: "in_review" }; + * export const hooks = { + * async onEvent({ event, actions }) { + * if (event.type === "parked" && event.reason === "max_visits_exhausted") { + * await actions.resume({ extraVisits: 1 }); // a scripted concierge + * } + * }, + * }; + * + * The returned spec goes through the full linter — scripting controls + * shape and management, never loosens the bounded-execution guarantees. + * Hooks run with operator authority (the same verbs a human concierge + * has), serialised off the event path; runaway hook loops are fenced by + * the same budgets and caps as everything else, so scripted flows should + * set a budget. + */ +import * as path from "node:path"; +import { pathToFileURL } from "node:url"; +import { mustLint } from "../spec/lint.js"; +import type { FlowSpec } from "../spec/types.js"; +import * as presets from "../presets.js"; +import { flow, FlowBuilder } from "./builder.js"; +import type { TicketState } from "../tracker/ticket.js"; +import type { FlowRun } from "../run/fold.js"; +import type { RunEvent } from "../run/events.js"; +import type { NudgeReceipt } from "../engine/ports.js"; +import type { ResumeGrant } from "../engine/stageManager.js"; + +/** What a flow script's default export receives. */ +export interface FlowScriptContext { + ticket: { + title: string; + body?: string; + criteria?: string[]; + }; + /** the fluent builder: flow().worker(...).build() */ + flow: typeof flow; + FlowBuilder: typeof FlowBuilder; + /** the shipped shapes, composable */ + presets: typeof presets; +} + +/** Everything a hook may do — a scripted concierge's operator verbs. */ +export interface HookActions { + nudge(text: string, node?: string): NudgeReceipt; + pause(): Promise; + resume(grant?: ResumeGrant): Promise; + decideGate(node: string, verdict: "pass" | "fail", note?: string): Promise; + cancel(reason?: string): Promise; + /** file follow-up work (needs may reference this ticket) */ + file(input: { + title: string; + body?: string; + criteria?: string[]; + needs?: string[]; + flow?: FlowSpec; + autoStaff?: boolean; + }): Promise<{ ref: string }>; +} + +export interface HookContext { + ref: string; + event: RunEvent; + run: FlowRun; + actions: HookActions; +} + +export interface FlowHooks { + /** invoked (serialised, off the event path) after every run event */ + onEvent?: (ctx: HookContext) => void | Promise; +} + +export interface LoadedFlow { + flow: FlowSpec; + stateMap?: Record; + hooks?: FlowHooks; + /** the resolved absolute path, stored on the ticket for recovery */ + scriptPath: string; +} + +type ScriptResult = + | FlowSpec + | { + flow: FlowSpec; + stateMap?: Record; + hooks?: FlowHooks; + }; + +interface FlowScriptModule { + default?: (ctx: FlowScriptContext) => ScriptResult | Promise; + stateMap?: Record; + hooks?: FlowHooks; +} + +/** + * Load and run a flow script. The returned flow is linted; a script that + * produces an unbounded or malformed graph fails at filing time exactly + * like a bad JSON spec would. Values returned from the function win over + * module-level exports. + */ +export async function loadFlowScript( + scriptPath: string, + ticket: FlowScriptContext["ticket"], +): Promise { + const resolved = path.resolve(scriptPath); + // cache-bust so edited scripts reload without restarting the tracker + const url = `${pathToFileURL(resolved).href}?t=${Date.now()}`; + let module: FlowScriptModule; + try { + module = (await import(url)) as FlowScriptModule; + } catch (err) { + throw new Error( + `flow script ${resolved} failed to load: ${err instanceof Error ? err.message : String(err)}`, + ); + } + // CJS interop: `module.exports = { default: fn, hooks }` arrives as the + // whole exports object in the namespace's default slot — unwrap it. + if ( + typeof module.default === "object" && + module.default !== null && + typeof (module.default as unknown as FlowScriptModule).default === "function" + ) { + module = module.default as unknown as FlowScriptModule; + } + if (typeof module.default !== "function") { + throw new Error(`flow script ${resolved} must default-export a function(ctx) → flow`); + } + const result = await module.default({ ticket, flow, FlowBuilder, presets }); + const isWrapped = result != null && typeof result === "object" && "flow" in result && !("version" in result); + const rawFlow = isWrapped ? (result as { flow: FlowSpec }).flow : (result as FlowSpec); + const linted = mustLint(rawFlow); + const stateMap = (isWrapped ? (result as { stateMap?: Record }).stateMap : undefined) ?? module.stateMap; + const hooks = (isWrapped ? (result as { hooks?: FlowHooks }).hooks : undefined) ?? module.hooks; + return { + flow: linted, + ...(stateMap !== undefined ? { stateMap } : {}), + ...(hooks !== undefined ? { hooks } : {}), + scriptPath: resolved, + }; +} diff --git a/src/cli.ts b/src/cli.ts index 2df3794..6b0ab7e 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -16,7 +16,8 @@ * against a server (--api http://127.0.0.1:PORT): * run-of-show board * run-of-show file --title T [--body B] [--criteria C]… [--needs R]… - * [--parent R] [--flow spec.json] [--no-auto-staff] + * [--parent R] [--flow spec.json | --flow-script flow.mjs] + * [--no-auto-staff] * run-of-show ticket * run-of-show staff * run-of-show nudge "text" [--node N] @@ -201,6 +202,9 @@ switch (command) { ...(arg("--parent") !== undefined ? { parent: arg("--parent") } : {}), ...(arg("--channel") !== undefined ? { originChannel: arg("--channel") } : {}), ...(flowFile ? { flow: JSON.parse(fs.readFileSync(flowFile, "utf8")) } : {}), + ...(arg("--flow-script") !== undefined + ? { flowScript: path.resolve(arg("--flow-script")!) } + : {}), ...(has("--no-auto-staff") ? { autoStaff: false } : {}), }).then((data) => console.log(JSON.stringify((data as { ticket: unknown }).ticket, null, 2))); break; diff --git a/src/index.ts b/src/index.ts index ffe1037..d9d275d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -36,5 +36,9 @@ export * from "./tracker/store.js"; export * from "./tracker/tracker.js"; export * from "./server.js"; +// Scripted authoring: the fluent builder + .js flow scripts with hooks +export * from "./authoring/builder.js"; +export * from "./authoring/script.js"; + // Appendix A + §8 presets export * as presets from "./presets.js"; diff --git a/src/tracker/ticket.ts b/src/tracker/ticket.ts index 61bddc5..a0d7f37 100644 --- a/src/tracker/ticket.ts +++ b/src/tracker/ticket.ts @@ -46,6 +46,8 @@ export interface Ticket { needs: string[]; /** the run of show; when absent, compiled from `cast` at staffing (§3.2) */ flow?: FlowSpec; + /** the flow was authored by this script (resolved path) — reloaded on boot */ + flowScript?: string; cast?: HarnessSpec; /** node → tracker column; falls back to node-kind defaults */ stateMap?: Record; @@ -69,6 +71,8 @@ export const fileTicketSchema = z parent: z.string().optional(), needs: z.array(z.string()).optional(), flow: flowSpecSchema.optional(), + /** author the flow with a .js/.mjs/.cjs script instead (see src/authoring) */ + flowScript: z.string().optional(), cast: harnessSpecSchema.optional(), stateMap: z.record(z.string(), ticketStateSchema).optional(), autoStaff: z.boolean().optional(), diff --git a/src/tracker/tracker.ts b/src/tracker/tracker.ts index b89aac0..a6bd348 100644 --- a/src/tracker/tracker.ts +++ b/src/tracker/tracker.ts @@ -34,6 +34,7 @@ import { type TicketState, } from "./ticket.js"; import { TicketStore } from "./store.js"; +import { loadFlowScript, type FlowHooks, type HookActions } from "../authoring/script.js"; const DEFAULT_CAST: HarnessSpec = { harness: "claude", model: "claude-sonnet-5", effort: "high" }; @@ -49,8 +50,10 @@ export interface TicketStatus { export class Tracker { readonly engine: StageManager; readonly tickets: TicketStore; - /** serialised async work (auto-staffing promoted tickets) */ + /** serialised async work (auto-staffing promoted tickets, script hooks) */ private work: Promise = Promise.resolve(); + /** per-ticket flow-script hooks — in-memory, reloaded on recover() */ + private hooks = new Map(); constructor( root: string, @@ -82,7 +85,24 @@ export class Tracker { async file(input: FileTicketInput): Promise { const parsed = fileTicketSchema.parse(input); + if (parsed.flow && parsed.flowScript) { + throw new Error("provide flow OR flowScript, not both"); + } if (parsed.flow) mustLint(parsed.flow); // fail at filing, not staffing + // Scripted authoring: run the script with the ticket context; the shape + // is computed for this task, then linted and frozen like any other spec. + let scriptHooks: FlowHooks | undefined; + if (parsed.flowScript) { + const loaded = await loadFlowScript(parsed.flowScript, { + title: parsed.title, + ...(parsed.body !== undefined ? { body: parsed.body } : {}), + ...(parsed.criteria !== undefined ? { criteria: parsed.criteria } : {}), + }); + parsed.flow = loaded.flow; + parsed.stateMap ??= loaded.stateMap; + parsed.flowScript = loaded.scriptPath; + scriptHooks = loaded.hooks; + } const now = new Date().toISOString(); const ticket = this.tickets.update((tasks): Ticket => { @@ -130,6 +150,7 @@ export class Tracker { children: [], needs, ...(parsed.flow !== undefined ? { flow: parsed.flow as FlowSpec } : {}), + ...(parsed.flowScript !== undefined ? { flowScript: parsed.flowScript } : {}), ...(parsed.cast !== undefined ? { cast: parsed.cast as HarnessSpec } : {}), ...(parsed.stateMap !== undefined ? { stateMap: parsed.stateMap as Record } @@ -142,6 +163,7 @@ export class Tracker { tasks.tickets[ref] = ticket; return ticket; }); + if (scriptHooks) this.hooks.set(ticket.ref, scriptHooks); if (ticket.state === "todo" && ticket.autoStaff) { await this.staff(ticket.ref); @@ -251,6 +273,26 @@ export class Tracker { /** Boot: replay runs, re-project every ticket, run missed promotions. */ async recover(): Promise { + // Reload flow-script hooks before replay so the scripted concierge + // hears the recovery events too. A script that fails to load is + // journalled and skipped — the run itself is unaffected. + for (const ticket of this.tickets.list()) { + if (!ticket.flowScript || isTerminal(ticket.state)) continue; + try { + const loaded = await loadFlowScript(ticket.flowScript, { + title: ticket.title, + ...(ticket.body !== undefined ? { body: ticket.body } : {}), + ...(ticket.criteria !== undefined ? { criteria: ticket.criteria } : {}), + }); + if (loaded.hooks) this.hooks.set(ticket.ref, loaded.hooks); + } catch (err) { + this.engine.store.journal( + ticket.ref, + new Date().toISOString(), + `flow-script reload failed: ${err instanceof Error ? err.message : String(err)}`, + ); + } + } await this.engine.recoverAll(); for (const ticket of this.tickets.list()) { if (ticket.staffed) this.reproject(ticket.ref); @@ -268,6 +310,51 @@ export class Tracker { if (!this.tickets.get(ref)) return; // runs opened outside the tracker this.reprojectFrom(ref, run); if (event.type === "run_done") this.promoteDependents(ref); + // Flow-script hooks: the scripted concierge sees every event, off the + // event path so its operator verbs never re-enter the engine mid-append. + const hooks = this.hooks.get(ref); + if (hooks?.onEvent) { + const onEvent = hooks.onEvent; + this.work = this.work.then(async () => { + try { + await onEvent({ ref, event, run, actions: this.hookActions(ref) }); + } catch (err) { + // hook failures are announced in the journal, never crash the run + this.engine.store.journal( + ref, + new Date().toISOString(), + `flow-script hook error on ${event.type}: ${err instanceof Error ? err.message : String(err)}`, + ); + } + }); + } + } + + private hookActions(ref: string): HookActions { + return { + nudge: (text, node) => this.nudge(ref, text, node), + pause: async () => { + await this.pause(ref); + }, + resume: async (grant) => { + await this.resume(ref, grant); + }, + decideGate: async (node, verdict, note) => { + await this.decideGate(ref, node, verdict, note); + }, + cancel: async (reason) => { + await this.cancel(ref, reason); + }, + file: async (input) => { + const ticket = await this.file(input as FileTicketInput); + return { ref: ticket.ref }; + }, + }; + } + + /** Tickets with live flow-script hooks (introspection, tests). */ + hookRefs(): string[] { + return [...this.hooks.keys()]; } private reproject(ref: string): void { diff --git a/test/authoring.test.ts b/test/authoring.test.ts new file mode 100644 index 0000000..7469d53 --- /dev/null +++ b/test/authoring.test.ts @@ -0,0 +1,341 @@ +/** + * Scripted workflow authoring: the fluent builder, .js flow scripts that + * compute shape per task, and hooks — the scripted concierge. + */ +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { describe, expect, it } from "vitest"; +import { + CollectingSink, + ManualClock, + SimulatedMergeProvider, + SimulatedSpawnAdapter, + Tracker, + flow, + loadFlowScript, +} from "../src/index.js"; + +function scriptFile(source: string, name = "test.flow.mjs"): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "ros-flowscript-")); + const file = path.join(dir, name); + fs.writeFileSync(file, source); + return file; +} + +interface Rig { + tracker: Tracker; + adapter: SimulatedSpawnAdapter; + root: string; +} + +function makeRig(root?: string): Rig { + const clock = new ManualClock(); + const adapter = new SimulatedSpawnAdapter(clock); + const dir = root ?? fs.mkdtempSync(path.join(os.tmpdir(), "ros-authoring-")); + const tracker = new Tracker(dir, adapter, new SimulatedMergeProvider(), new CollectingSink(), { + clock, + ownerDM: "dm", + }); + adapter.connect((ref, seatKey, ev) => tracker.engine.deliverWorkerEvent(ref, seatKey, ev)); + return { tracker, adapter, root: dir }; +} + +describe("the flow builder", () => { + it("composes the reviewed lifecycle fluently, first node as entry, park defaults", () => { + const spec = flow() + .worker("implement", { + cast: { harness: "pi", effort: "high" }, + onPass: "review", + maxVisits: 3, + }) + .gate("review", { + by: { cast: { harness: "claude", effort: "high" }, rubric: "criteria-vs-diff" }, + onPass: "done", + onFail: "implement", + maxFails: 3, + }) + .budget({ usd: 10 }) + .build(); + expect(spec.entry).toBe("implement"); + expect(spec.nodes["implement"]).toMatchObject({ kind: "worker", onFail: "park" }); + expect(spec.budget).toEqual({ usd: 10 }); + }); + + it("builds fanout/join shapes with sane defaults", () => { + const spec = flow() + .fanout("split", { + arms: [{ cast: { harness: "pi" } }, { cast: { harness: "codex" } }], + join: "land", + }) + .join("land", { strategy: "all-merge", onPass: "done" }) + .build(); + expect(spec.nodes["split"]).toMatchObject({ isolation: "worktree-each" }); + expect(spec.nodes["land"]).toMatchObject({ onFail: "park" }); + }); + + it("refuses duplicate nodes, empty flows, and lint-invalid graphs at build()", () => { + expect(() => + flow() + .worker("a", { cast: { harness: "pi" }, onPass: "done" }) + .worker("a", { cast: { harness: "pi" }, onPass: "done" }), + ).toThrow(/already defined/); + expect(() => flow().build()).toThrow(/no entry/); + expect(() => + flow().worker("a", { cast: { harness: "pi" }, onPass: "ghost" }).build(), + ).toThrow(/unknown_edge_target/); + }); +}); + +describe("loadFlowScript", () => { + it("runs the default export with the ticket context and lints the result", async () => { + const file = scriptFile(` + export default ({ ticket, flow }) => + flow().worker("implement", { + cast: { harness: "pi", effort: ticket.title.includes("big") ? "high" : "low" }, + onPass: "done", + }).build(); + `); + const small = await loadFlowScript(file, { title: "small tweak" }); + const big = await loadFlowScript(file, { title: "big refactor" }); + expect((small.flow.nodes["implement"] as { cast: { effort: string } }).cast.effort).toBe("low"); + expect((big.flow.nodes["implement"] as { cast: { effort: string } }).cast.effort).toBe("high"); + expect(small.scriptPath).toBe(file); + }); + + it("supports async scripts, presets in context, wrapped returns and module exports", async () => { + const file = scriptFile(` + export default async ({ presets }) => ({ + flow: presets.onePass(), + stateMap: { implement: "in_progress" }, + hooks: { onEvent: () => {} }, + }); + `); + const loaded = await loadFlowScript(file, { title: "t" }); + expect(Object.keys(loaded.flow.nodes)).toEqual(["implement"]); + expect(loaded.stateMap).toEqual({ implement: "in_progress" }); + expect(loaded.hooks?.onEvent).toBeTypeOf("function"); + + const moduleLevel = scriptFile(` + export default ({ presets }) => presets.onePass(); + export const stateMap = { implement: "in_review" }; + export const hooks = { onEvent: () => {} }; + `); + const loaded2 = await loadFlowScript(moduleLevel, { title: "t" }); + expect(loaded2.stateMap).toEqual({ implement: "in_review" }); + expect(loaded2.hooks?.onEvent).toBeTypeOf("function"); + }); + + it("works with CommonJS scripts too", async () => { + const file = scriptFile( + `module.exports = { default: ({ presets }) => presets.onePass() };`, + "test.flow.cjs", + ); + const loaded = await loadFlowScript(file, { title: "t" }); + expect(loaded.flow.entry).toBe("implement"); + }); + + it("fails loudly on missing exports, load errors and lint-invalid results", async () => { + await expect( + loadFlowScript(scriptFile(`export const x = 1;`), { title: "t" }), + ).rejects.toThrow(/must default-export a function/); + await expect( + loadFlowScript(scriptFile(`syntax error here(`), { title: "t" }), + ).rejects.toThrow(/failed to load/); + await expect( + loadFlowScript( + scriptFile(`export default () => ({ version: 1, entry: "ghost", nodes: {} });`), + { title: "t" }, + ), + ).rejects.toThrow(/bad_entry/); + }); +}); + +describe("scripted filing through the tracker", () => { + it("the same script produces different shapes for different tickets", async () => { + const rig = makeRig(); + const script = path.resolve("examples/flows/adaptive.flow.mjs"); + + const hotfix = await rig.tracker.file({ + title: "hotfix: broken link", + body: "fix it", + flowScript: script, + }); + const hotfixRun = rig.tracker.engine.status(hotfix.ref); + expect(Object.keys(hotfixRun.spec.nodes)).toEqual(["implement"]); + expect(hotfixRun.spec.budget).toEqual({ usd: 2 }); + + const feature = await rig.tracker.file({ + title: "notification prefs", + body: "areas: api, ui, docs", + flowScript: script, + }); + const featureRun = rig.tracker.engine.status(feature.ref); + expect(Object.keys(featureRun.spec.nodes).sort()).toEqual(["land", "review", "split"]); + const split = featureRun.spec.nodes["split"] as { arms: Array<{ brief: string }> }; + expect(split.arms).toHaveLength(3); + expect(split.arms[1]!.brief).toContain("ui"); + // the script's flow was linted and frozen; the ticket remembers its author + expect(rig.tracker.get(feature.ref).flowScript).toBe(script); + expect(rig.tracker.get(feature.ref).flow).toBeTruthy(); + }); + + it("refuses flow + flowScript together", async () => { + const rig = makeRig(); + await expect( + rig.tracker.file({ + title: "t", + flow: { version: 1, entry: "implement", nodes: { implement: { kind: "worker", cast: { harness: "pi" }, onPass: "done", onFail: "park" } } }, + flowScript: "x.mjs", + }), + ).rejects.toThrow(/not both/); + }); + + it("a broken script fails the filing and stores nothing", async () => { + const rig = makeRig(); + const bad = scriptFile(`export default () => ({ version: 1, entry: "x", nodes: {} });`); + await expect(rig.tracker.file({ title: "t", flowScript: bad })).rejects.toThrow(/bad_entry/); + expect(rig.tracker.list()).toHaveLength(0); + }); +}); + +describe("hooks — the scripted concierge", () => { + it("auto-approves a human gate per the script's policy", async () => { + const rig = makeRig(); + const script = scriptFile(` + export default ({ flow }) => + flow() + .worker("implement", { cast: { harness: "pi", effort: "low" }, onPass: "approve" }) + .gate("approve", { by: "human", onPass: "done", onFail: "implement" }) + .build(); + export const hooks = { + async onEvent({ event, actions }) { + if (event.type === "parked" && event.reason === "human_gate") { + await actions.decideGate("approve", "pass", "auto-approved: low-risk policy"); + } + }, + }; + `); + const t = await rig.tracker.file({ title: "low-risk chore", flowScript: script }); + const seat = rig.adapter.seat("implement"); + await seat.ready(); + await seat.complete(); + await rig.tracker.settle(); // the hook fires off the event path + expect(rig.tracker.get(t.ref).state).toBe("done"); + const decided = rig.tracker.engine.store + .readEvents(t.ref) + .find((e) => e.type === "gate_decided"); + expect(decided).toMatchObject({ by: "human", verdict: "pass", note: "auto-approved: low-risk policy" }); + }); + + it("auto-grants one extra visit when the rework loop parks (adaptive example)", async () => { + const rig = makeRig(); + const t = await rig.tracker.file({ + title: "stubborn feature", + body: "plain", + criteria: ["a", "b", "c"], // → reviewed lifecycle, high effort + flowScript: path.resolve("examples/flows/adaptive.flow.mjs"), + }); + // burn all three rework cycles + for (let visit = 1; visit <= 3; visit++) { + const seat = rig.adapter.seat("implement", { visit }); + await seat.ready(); + await seat.complete(); + if (visit < 3) { + const review = rig.adapter.seat("review", { visit }); + await review.ready(); + await review.complete({ data: { pass: false } }); + } + } + const review3 = rig.adapter.seat("review", { visit: 3 }); + await review3.ready(); + await review3.complete({ data: { pass: false } }); + // the park fires; the hook resumes with +1 visit + await rig.tracker.settle(); + const run = rig.tracker.engine.status(t.ref); + expect(run.status).toBe("running"); + expect(run.visits["implement"]).toBe(4); + const resumed = rig.tracker.engine.store.readEvents(t.ref).find((e) => e.type === "resumed"); + expect(resumed).toMatchObject({ grant: { extraVisits: 1 } }); + }); + + it("a hook can file follow-up work that depends on this ticket", async () => { + const rig = makeRig(); + const script = scriptFile(` + export default ({ presets }) => presets.onePass(); + export const hooks = { + async onEvent({ ref, event, actions }) { + if (event.type === "run_done") { + await actions.file({ title: "follow-up: docs for " + ref, autoStaff: false }); + } + }, + }; + `); + const t = await rig.tracker.file({ title: "main work", flowScript: script }); + const seat = rig.adapter.seat("implement"); + await seat.ready(); + await seat.complete(); + await rig.tracker.settle(); + const followUp = rig.tracker.list().find((x) => x.title.startsWith("follow-up")); + expect(followUp).toBeTruthy(); + expect(followUp!.title).toContain(t.ref); + expect(followUp!.state).toBe("todo"); + }); + + it("hook errors are journalled, never crash the run", async () => { + const rig = makeRig(); + const script = scriptFile(` + export default ({ presets }) => presets.onePass(); + export const hooks = { + onEvent({ event }) { + if (event.type === "run_done") throw new Error("policy engine exploded"); + }, + }; + `); + const t = await rig.tracker.file({ title: "t", flowScript: script }); + const seat = rig.adapter.seat("implement"); + await seat.ready(); + await seat.complete(); + await rig.tracker.settle(); + expect(rig.tracker.get(t.ref).state).toBe("done"); // unharmed + expect( + rig.tracker.engine.store.readJournal(t.ref).some((l) => l.includes("policy engine exploded")), + ).toBe(true); + }); + + it("recovery reloads hooks from the stored script path", async () => { + const rig = makeRig(); + const script = scriptFile(` + export default ({ flow }) => + flow() + .worker("implement", { cast: { harness: "pi", effort: "low" }, onPass: "approve" }) + .gate("approve", { by: "human", onPass: "done", onFail: "implement" }) + .build(); + export const hooks = { + async onEvent({ event, actions }) { + if (event.type === "parked" && event.reason === "human_gate") { + await actions.decideGate("approve", "pass", "auto-approved after reboot"); + } + }, + }; + `); + const t = await rig.tracker.file({ title: "survives reboots", flowScript: script }); + const seat = rig.adapter.seat("implement"); + await seat.ready(); + + // reboot before the gate is reached + const rebooted = makeRig(rig.root); + await rebooted.tracker.recover(); + expect(rebooted.tracker.hookRefs()).toContain(t.ref); + // recovery re-staffed the seat; drive it to the gate — the reloaded hook approves + const seat2 = rebooted.adapter.seat("implement", { attempt: 2 }); + await seat2.ready(); + await seat2.complete(); + await rebooted.tracker.settle(); + expect(rebooted.tracker.get(t.ref).state).toBe("done"); + const decided = rebooted.tracker.engine.store + .readEvents(t.ref) + .find((e) => e.type === "gate_decided"); + expect(decided).toMatchObject({ note: "auto-approved after reboot" }); + }); +});