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..20ef73e 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,89 @@ 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. + +## 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 @@ -137,6 +216,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/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/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/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 c4907d8..6b0ab7e 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -1,13 +1,28 @@ #!/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 | --flow-script flow.mjs] + * [--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 +30,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 +92,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 +149,116 @@ 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")) } : {}), + ...(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; + } + 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..d9d275d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -28,6 +28,17 @@ 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"; + +// 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/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..a0d7f37 --- /dev/null +++ b/src/tracker/ticket.ts @@ -0,0 +1,82 @@ +/** + * 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; + /** 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; + /** 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(), + /** 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(), + }) + .strict(); + +export type FileTicketInput = z.infer; diff --git a/src/tracker/tracker.ts b/src/tracker/tracker.ts new file mode 100644 index 0000000..a6bd348 --- /dev/null +++ b/src/tracker/tracker.ts @@ -0,0 +1,463 @@ +/** + * 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"; +import { loadFlowScript, type FlowHooks, type HookActions } from "../authoring/script.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, script hooks) */ + private work: Promise = Promise.resolve(); + /** per-ticket flow-script hooks — in-memory, reloaded on recover() */ + private hooks = new Map(); + + 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 && 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 => { + // 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.flowScript !== undefined ? { flowScript: parsed.flowScript } : {}), + ...(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 (scriptHooks) this.hooks.set(ticket.ref, scriptHooks); + + 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 { + // 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); + } + // 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); + // 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 { + 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/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" }); + }); +}); 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"); + }); +});