// queue-store.ts — the programmatic orchestrator queue store. Extension-owned, // machine-readable, with free-form text fields (notes/description carry the // schema-free content). state.md becomes a RENDER of this store. // // Statuses: proposal → approved → active → ai-review → human-review → done // ↘ rejected ↘ failed ↗ (re-dispatch) // ↘ failed ←(recovery)→ active // ↘ blocked (PARKED: held for a human, no auto-recovery) // ai-review → no-change is the NON-DELIVERABLE terminal outcome (AUTOPILOT-108): // the reviewer ruled the correct thing was NO change — never auto-re-dispatched. // // Ownership split: approval/dispatch/review verdicts are orchestrator // judgment (via queue_* tools); active→ai-review/failed are extension events // (async-complete). Only the two event transitions happen automatically. import { readFileSync, writeFileSync, existsSync, renameSync, mkdirSync, linkSync, statSync, unlinkSync } from "node:fs"; import { join } from "node:path"; import { hostname } from "node:os"; import { randomBytes } from "node:crypto"; import { isValidStatus, validTransition, type QueueStatus } from "./framework/item-contract.ts"; import type { ReviewNotes } from "./review-notes.ts"; import type { StoredAcceptanceEvidence } from "./acceptance-evidence.ts"; // The status union, the status parser, and the transition map live in the // item-contract module (AUTOPILOT-128) — the ONE source of truth the write // validation, the documented contract text and the operator surface share. // Re-exported here so every existing `from "./queue-store.ts"` import keeps // working. export { validTransition }; export { isValidStatus }; export type { QueueStatus }; // `no-change` (KEY: AUTOPILOT-108) is the NON-DELIVERABLE terminal status: the // reviewer ruled that the correct outcome was to make NO change (a // refusal/withdrawal, no artifact). It is NOT a rejection on quality and it is // NEVER auto-re-dispatched (auto-dispatch only touches `approved`, auto-recovery // only `failed`); see verdict.ts's NO-CHANGE and core.ts's routing. export type BlockerReason = "parked" | "serialized" | "merge" | "decision" | null; /** The status an item was PARKED FROM (KEY: AUTOPILOT-109), recorded on every * transition INTO `blocked`. The dependency-release sweep restores THIS status * when the item's dependencies complete, instead of unconditionally * `approved` — so an automatic release can never promote a park that was never * approved (a park from `proposal` must return to `proposal` and still face * the human approval gate). `null`/absent on a pre-AUTOPILOT-109 record, and * the release refuses to guess for those (no origin = no auto-approval). */ export type BlockedFrom = "approved" | "proposal" | "failed" | "human-review"; /** WHY an item reached `failed` — the failed=cap vs failed=verdict distinction * (BUDGET-GOVERNANCE). `budget-capped`: the worker run was CUT OFF by the * item's requested wall-clock budget (timeoutMs) mid-task — the failure says * nothing about the work, and the re-dispatch must run with a BIGGER budget. * `verdict`: the run ended with an unsuccessful verdict/exit (incl. a review * FAIL reaching the attempts cap). `zombie`: the run's completion event was * lost (fleet idle past grace). `spawn`: the provider/infra layer failed the * run — rejected AT SPAWN (bare 400) or died mid-run producing NO deliverable * (empty api_error) — so the failure says nothing about the work and the * recovery is a bounded provider retry (2× then escalate, AUTO-RECOVER-FAILS). * null = not failed, or a legacy/pre-governance failure with no recorded * cause. */ export type FailCause = "budget-capped" | "verdict" | "zombie" | "spawn"; export function isFailCause(v: unknown): v is FailCause { return v === "budget-capped" || v === "verdict" || v === "zombie" || v === "spawn"; } /** WHERE a dispatched run writes (FINISHER-EVIDENCE). `worker` (default): the * run writes ONLY inside its isolated worktree, so worktree edits are the * evidence of work. `finisher`: the run writes OUTSIDE its worktree — into * the DECLARED cwd's checkout (a merge finisher cherry-picks/merges an * approved branch into the target repo's main checkout) — so its worktree is * untouched BY DESIGN and a runtime 'no edits' heuristic misreads it as * "returned planning output". For that class the queue's own record is * authoritative: the evidence is the DECLARED SOURCE entering the declared * cwd's history during the run (`landedEvidence`), not worktree file edits * — and not a bare HEAD move, which other lanes and humans also produce. */ export type DispatchClass = "worker" | "finisher"; export function isDispatchClass(v: unknown): v is DispatchClass { return v === "worker" || v === "finisher"; } /** The declared-cwd HEAD as it stood WHEN a finisher-class run was dispatched, * plus the SOURCE that run was sent to land — together, the baseline the * landed check compares against. Written by EVERY lane that spawns a * finisher-class run, and cleared when the item (re-)enters `active`, so a * baseline always describes the item's most recent dispatch. */ export interface FinisherBaseline { /** The checkout the finisher writes to (the item's declared cwd). */ repo: string; /** The checked-out ref at dispatch (e.g. `main`), or null when detached. */ ref: string | null; /** HEAD sha at dispatch; null when the repo could not be read. */ sha: string | null; /** WHAT this dispatch was sent to land (branch/tag/sha), or null when the * dispatch declared none — with no source there is NO landed evidence. */ source: string | null; /** `source` resolved to a commit AT DISPATCH, so a branch deleted by the * merge can still answer "is it in the target's history now?". */ sourceSha: string | null; /** The run this baseline belongs to. */ runId: string | null; at: string; } /** PROOF that a finisher-class dispatch's work LANDED: the source it was sent * to land entered the declared cwd's history during the run. This is the * success evidence for work that never touches its own worktree. */ /** How a declared source is present in a history: `ancestor` = the very * commits it names (merge/fast-forward); `patch-equivalent` = copies carrying * the same patch (cherry-pick/rebase/squash of a single-commit source). */ export type LandingShape = "ancestor" | "patch-equivalent"; export interface LandedEvidence { repo: string; ref: string | null; /** HEAD at dispatch (the baseline) — null when it could not be read. */ fromSha: string | null; /** HEAD after the run — the landed commit. */ sha: string; /** HOW the source is present. Absent on records written before the shape * was tracked. */ landing?: LandingShape; /** The source that landed, as declared at dispatch, and its commit. */ source: string | null; sourceSha: string; /** The run whose work this is. */ runId: string | null; at: string; } /** A RECORDED override of a run-level failure verdict — the thing that used to * be hand-written prose in `notes`. Machine-readable so a PATTERN of * overrides is visible (many overrides = the runtime verdict is systematically * wrong, or the operator is waving failures through). */ export interface FailureOverride { at: string; /** `framework` = the harness overrode a runtime verdict against recorded * evidence; `orchestrator` = a human/agent judgment call. */ by: "framework" | "orchestrator"; /** The run whose failure verdict was overridden. */ runId: string | null; /** WHY the failure was not believed. */ reason: string; /** The landed evidence that justified a framework override (null for a * judgment-only orchestrator override). */ evidence: LandedEvidence | null; } /** The TERMINAL statuses — an item in one of these is CLOSED: it owns no live * run, expects no further write, and is NEVER auto-re-dispatched. * - `done` = human-approved and complete; * - `rejected` = deliberately dropped; * - `no-change` = the non-deliverable terminal outcome (AUTOPILOT-108): the * reviewer ruled the correct outcome was NO change. * This is the ONE closed set every terminal-status branch must read, so a new * terminal member is added HERE and nowhere else. The recurring bug this * guards against is a closed set grown at one call site and missed at the * others (AUTOPILOT-108 added `no-change` but left `preDoneKeysFor` and the * keep-ref retention behind). A caller that genuinely needs a DIFFERENT set * (e.g. keep-ref retention prunes on `done`/`rejected` but deliberately * keeps, then merge-gates, a `no-change` ref) must say so explicitly there. */ export const TERMINAL_STATUSES: ReadonlySet = new Set(["done", "rejected", "no-change"]); /** Is `status` terminal (see `TERMINAL_STATUSES`)? The single predicate every * terminal-status branch shares. */ export function isTerminalStatus(status: QueueStatus): boolean { return TERMINAL_STATUSES.has(status); } export interface QueueItem { key: string; status: QueueStatus; /** approved = dispatchable; blocked = approved-but-waiting (blocker says why). */ blocker: BlockerReason; /** WHERE this item was parked from (see BlockedFrom). Set on every transition * INTO `blocked`; the release sweep restores it. Cleared on the way OUT of * `blocked` (a park is over). null/absent = never blocked, or a legacy park * with no recorded origin. */ blockedFrom?: BlockedFrom | null; title: string; /** The worker-prompt scope (auto-dispatch builds the task from this). */ scope: string; /** The repo the worker runs in (auto-dispatch requires it). */ cwd: string | null; evidence: string; /** CLOSED set H|M|L (item-contract.ts VALUE_VALUES) — required at approval, * refused when outside the set. Not free-form prose. */ value: string; /** CLOSED set H|M|L (item-contract.ts URGENCY_VALUES) — see `value`. */ urgency: string; /** CLOSED set low|medium|high (item-contract.ts RISK_VALUES) — required at * approval AND dispatch; a prose value (`low — behaviour-preserving`) or a * non-vocabulary token (`med`, `low-risk`) is REFUSED at the transition, not * stored and silently skipped. The rationale belongs in evidence/notes. */ risk: string; runId: string | null; /** The last run id `runId` held before a terminal transition (done/ * human-review/failed/no-change) nulled it (KEY: AUTOPILOT-85, set by updateItem's * terminal-transition clearing, below). `runId` says which run is CURRENTLY * live; once a run reaches a terminal state there is no live run, but * `resolveWorkTip`'s runWorktree gate (shipping.ts) still needs the identity * of the run that PRODUCED the terminal state to tell a genuine tip apart * from a superseded attempt's stale worktree record. Sticky across * re-statements of a terminal status (never re-cleared by re-entering the * SAME terminal status) and only overwritten by the NEXT terminal * transition's own clearing. Dispatch (entering `active`) does not touch it * — only the terminal-transition clearing in updateItem writes it. Absent = * a pre-AUTOPILOT-85 item, or one that has never reached a terminal status * with a live runId to record. Deliberately NOT backfilled on `loadStore` * (an old item genuinely never had this fact recorded, and absent/undefined * reads exactly like null everywhere it is checked, so there is nothing for * a backfill to normalize). */ lastRunId?: string | null; /** reviewer run id when a queue_review was spawned for this item */ reviewerRunId: string | null; /** SPEC PASS (KEY: AUTOPILOT-36) — the run id of a bounded, READ-ONLY * reconnaissance worker dispatched against THIS proposal (panel action * `spec-pass`) to complete its scope. Scoped to `proposal` items only — * see `itemBySpecPassRunId`/updateItem's leaving-proposal stamping below. * null = no spec pass in flight. Setting/clearing this NEVER moves status: * a spec pass never approves anything, and approval stays human-only. * OPTIONAL (unlike `reviewerRunId`) — deliberately, so this newer field * does not force every existing QueueItem literal across the codebase * (tests especially) to be touched; absent/undefined reads exactly like * null everywhere this module and its callers check it, and `loadStore` * still backfills it below for anything persisted to disk. */ specPassRunId?: string | null; /** SPEC PASS pending marker: set by the panel's `spec-pass` action (never * a status move — see `PanelActionId`), consumed by the harness's * sweep-time dispatcher (`dispatchPendingSpecPass`, auto-dispatch.ts) * which claims it (sets false) before spawning — the SAME claim-before- * spawn discipline `pendingRedispatchFindings` uses (AUTOPILOT-59) * against a double dispatch. Absent/false = nothing pending. */ pendingSpecPass?: boolean; /** Requested wall-clock budget (ms) recorded on the item — EVERY dispatch * lane (manual dispatch/review + auto-dispatch/re-dispatch/auto-review) * passes it to backend.spawn so the child inherits the requested budget * instead of a runtime default. null = unset → runtime default applies. */ timeoutMs: number | null; /** MODEL OVERRIDE (KEY: AUTOPILOT-79, reviewer lanes added at AUTOPILOT-80): * `provider/id[:thinking]` passed to backend.spawn by every WORKER lane * (manual dispatch, auto-dispatch, human re-dispatch, spec pass, * auto-recovery) AND every REVIEWER lane (autoReview, queue_review) — * resolveSpawnModel (spawn-model.ts) resolves it there ahead of * autopilot.config.json's config-wide `model` default. Absent/undefined = * falls through to the config default, else the spawned child inherits * the session's model exactly as before either feature existed — the * backends add nothing to the launch when nothing resolves. Not validated * against the model registry. */ model?: string; /** CAPABILITY LEVEL (KEY: AUTOPILOT-110/111): a level NAME (e.g. high|medium| * low) resolved through the level map at DISPATCH time — ONE place to * re-point the user's model choices, replacing a hardcoded vendor id in the * item scope. The map is ACTUATOR-owned (`actuator.config.json` → `models`); * autopilot.config.json's own `models` is only a fallback (see config.ts). A * raw `model` (above) still beats it; it beats the config default tier. * Shape-validated (non-empty) at write; an unknown level FAILS LOUDLY at * resolution (resolveSpawnModel), never silently falling back to inherit. * Absent = no tier. */ tier?: string; /** re-dispatch attempt counter (review-FAIL cap is 5) */ attempts: number; /** DEPENDENCY GATE (KEY: AUTOPILOT-81): keys of OTHER items this one is * BLOCKED ON until they reach `done` — `done` is the release condition, * NOT human-review and NOT `shippedAt` (shipping can be a deliberate, * later, manual step). Validated at every write (`validateAfter`): every * key must already exist in the store, no self-reference, no cycle — a bad * `after` is rejected with a clear reason instead of silently stored. An * `approved` item with unmet `after` deps is WAITING, not ready (see * `unmetDependencies`): every dispatch lane (auto-dispatch eligibility, * autoRedispatch/human re-dispatch, spec-pass dispatch, auto-recovery's * candidate filter, queue_dispatch) skips or refuses it — `overrideCapacity` * does NOT override this; the only way past it is editing `after`. * Absent/empty = no dependencies. */ after?: string[]; /** SHIPPING MARKER (KEY: AUTO-SHIP-ON-DONE): the ISO timestamp when the * merge-finisher shipped this item (merged into main / MRs created). * absent/null = not shipped yet — a `done` item without this marker is * still waiting for the shipping lane. The marker prevents re-merge: * once set, the shipping lane skips the item. Cleared when the item * leaves `done` (a human re-open starts a fresh approval + ship cycle). */ shippedAt?: string | null; /** Why this item reached `failed` (see FailCause). null when not failed or * a legacy failure with no recorded cause. Set by the event-driven flips * (worker completion, review-cap, zombie reconciliation) and cleared on any * transition OUT of failed — a stale cause must never ride into a fresh * run. The capped-failure path ALSO writes a human note, so both the * machine flag and the operator-visible text say 'budget-capped'. */ failCause?: FailCause | null; /** AUTO-RECOVER-FAILS bookkeeping: how many automatic recovery attempts * this item has consumed (spawn calls made by the recovery engine). Bounded * by the recovery policy per cause + the global cap (default 2) — an item * that exhausts them stays `failed` with a one-time escalation tick. Counts * FAILED attempts too (a spawn the provider rejected consumed an attempt). */ recoveries?: number; /** Epoch ms before which the next automatic recovery attempt must NOT run * (the short backoff after a failure). null = eligible on the next pass. */ recoveryNotBefore?: number | null; /** true once the recovery budget is spent and the escalation tick fired — * the item stays failed for the orchestrator/human to act on. Prevents a * re-escalation tick on every sweep. */ recoveryEscalated?: boolean; /** WHERE this item's dispatched run writes (see DispatchClass). Absent = * `worker` (writes only inside its worktree). `finisher` declares that the * run writes into the DECLARED cwd's checkout, so an empty worktree is * expected and NEVER evidence of a run that did nothing. */ dispatchClass?: DispatchClass; /** WHAT a finisher-class dispatch lands (branch/tag/sha). Sticky across * re-dispatches — it describes the item's work, not one run — and required * for landed evidence (no declared source → no evidence). */ finisherSource?: string | null; /** The declared cwd's HEAD when the current finisher run was dispatched. */ finisherBaseline?: FinisherBaseline | null; /** Evidence that the work LANDED in the declared cwd (the declared source * entered its history). Sticky: it records a fact about this item's work, * and auto-recovery refuses to re-dispatch an item that carries it (a * re-run would duplicate a merge that already landed). */ landedEvidence?: LandedEvidence | null; /** WHERE the current run's managed worktree lives on disk (AUTOPILOT-47 C). * The runtime creates `pi-parallel--*` in a worktree it is documented * to later delete (AUTOPILOT-74 found that deletion is not reliable — see * worktree-preservation.ts's `pruneStaleWorktrees`); * when a run is reaped, UNCOMMITTED changes survive in that directory but * nothing recorded which directory it was, so salvage meant matching run ids * to `pi-worktree-*` paths by hand. Resolved from `git worktree list` during * the preservation pass (the path does not exist yet at dispatch) and left in * place after the run ends — a stale path is the post-mortem's starting * point, not a lie about a live run (`runId` says which run it belonged to). * Absent = never resolved (no matching worktree, or a non-worktree run). */ runWorktree?: { runId: string; path: string; branch: string } | null; /** Recorded overrides of run-level failure verdicts (append-only). */ overrides?: FailureOverride[]; /** HUMAN RE-DISPATCH (KEY: AUTOPILOT-59, second producer KEY: AUTOPILOT-69): * the findings recorded for a human-review item's re-dispatch, NOT YET * spawned. Set in the SAME mutateStore write that moves the item * human-review → active — from EITHER of two producers: the decision * panel's "redispatch" action (panels.ts, a human-operated host UI) or * queue_update's `findings` param (queue-ops.ts, the tool surface the * orchestrator itself calls — it has no panel to click). Both write the * identical marker; the harness's sweep-time consumer * (dispatchPendingHumanRedispatch, auto-dispatch.ts) reads it regardless of * which producer wrote it, spawns the redo via workerTask(item, findings) * (the SAME findings-carrying dispatcher a review FAIL uses), and clears * it. null/absent = nothing pending. Distinct from the notes-appended copy * (durable, human-readable audit trail): this field is the machine-readable * handoff, cleared once consumed so a completed dispatch is never re-read * as still-pending. */ pendingRedispatchFindings?: string | null; /** THE WORKER'S ACCEPTANCE ARTIFACT (KEY: AUTOPILOT-108), captured from the * worker's completion at active → ai-review and BOUNDED at write time (see * acceptance-evidence.ts's cap — queue.json is re-read by every tick and * panel render). This is the auditable test evidence the reviewer AUDITS * instead of re-running the acceptance suite. null/absent = none was * captured (the reviewer must then record the load-bearing claims as COULD * NOT CHECK). Backfilled on read. */ acceptanceEvidence?: StoredAcceptanceEvidence | null; /** WHAT THE REVIEWER SAID BEYOND THE VERDICT (AUTOPILOT-27), captured at * verdict parse and BOUNDED at write time (see review-notes.ts caps — * queue.json is re-read by every tick and panel render). The approval * screen shows it, so the human decides with the information the review * produced instead of a bare "PASSED". null = nothing detectable (or a * pre-AUTOPILOT-27 record); backfilled on read. Overwritten by the next * verdict for this item — it describes the LATEST review, not a history. */ reviewNotes?: ReviewNotes | null; /** WHERE to review THIS attempt (AUTOPILOT-57): the same shape/semantics as * flag_for_review's `review_targets` — branch+commits (what to diff), * file:line anchors the reviewer cited as evidence, doc paths the work * changed, and an MR/PR link where a remote exists. Built by * `buildReviewTargets` (framework/review-targets.ts) at verdict PASS and * stored here so the handover survives a session, a reload and a scroll — * a rendered tick message is not durable, the item is. REFRESH SEMANTICS: * a full overwrite on every PASS (never appended) — a re-dispatched * attempt's stale branch/SHA is superseded, not stacked alongside the new * one. Empty array = built, nothing resolved (say so, never fabricate); * null/absent = never built (pre-AUTOPILOT-57 item, or no PASS yet). */ reviewTargets?: string[] | null; /** free-form notes/description — no schema constraints on content */ notes: string; createdAt: string; updatedAt: string; /** true when the key was allocated PROVISIONALLY (proposal with no cwd yet → * default Q series); a later approved transition with cwd RENAMES the key * into the repo's real series. Never set for explicit keys or deliberate * series choices. */ provisionalKey?: boolean; } export interface QueueStore { version: 1; /** Monotonic revision — half of the COMPARE-AND-SWAP token. Bumped by every * mutateStore write; a mutation that finds a different rev on disk than the * one it read has been overtaken and re-applies instead of clobbering. * Backfilled to 0 on read for pre-rev stores (no schema break). */ rev: number; /** The WRITE IDENTITY that stamped `rev` — the other half of the CAS token. * `rev` alone is not identifying: two writers that both read revision N both * stamp N+1, so each one's post-write re-read is satisfied by the OTHER's * write and both report success while one is silently gone. The nonce is * unique per write (`@:`), so a writer recognises its own * write and nothing else. Backfilled to null on read. */ revBy?: string | null; items: Record; } export type QueueLengths = Record; // --------------------------------------------------------------------------- // Transition rules // --------------------------------------------------------------------------- // The transition map (STATUS_TRANSITIONS) and validTransition live in // framework/item-contract.ts — the ONE source the contract text is generated // from; they are imported/re-exported at the top of this file. /** EVIDENCE OF A COMPLETED DELIVERABLE (KEY: AUTOPILOT-93) — the gate on the * `blocked → ai-review` transition. Returns a short human-readable description * of the evidence, or null when the item carries none. * * WHAT COUNTS, EXACTLY, and WHERE IT IS READ FROM — every source is the item's * OWN persisted record, so this predicate is PURE (no backend call, no * state-dir read): * * `runId` — a run id attached to the item (a recorded run). This is * the shape the normal completion path leaves: active → * ai-review KEEPS `runId` (only done/human-review/failed/ * no-change clear it). The tool route records the completed run * here explicitly (queue_update's `completedRun`). * * `lastRunId` — stamped by updateItem when a run drove this item to a * terminal lifecycle state (done/human-review/failed/no-change), * i.e. the framework's own record that a run for this * item RAN TO COMPLETION even after `runId` was cleared. * * `landedEvidence`— a landed commit recorded by the finisher-evidence lane * (a recorded SHA). * * `runWorktree` — a branch/worktree recorded by worktree-preservation for * a run (a recorded BRANCH). * * `shippedAt` — the shipping lane's marker. * * An item carrying NONE of these has no run that produced anything and must * NOT be reachable as `ai-review` by this route — that would turn a PARK into a * way to mark unfinished work reviewable. */ export function completedDeliverableEvidence( item: Pick, ): string | null { const runId = item.runId?.trim(); if (runId) return `run ${runId.slice(0, 8)}`; const lastRunId = item.lastRunId?.trim(); if (lastRunId) return `completed run ${lastRunId.slice(0, 8)}`; const landedSha = item.landedEvidence?.sha?.trim(); if (landedSha) return `landed ${landedSha.slice(0, 8)}`; const branch = item.runWorktree?.branch?.trim(); if (branch) return `branch ${branch}`; const shippedAt = item.shippedAt?.trim(); if (shippedAt) return `shipped ${shippedAt}`; return null; } // --------------------------------------------------------------------------- // Store IO // --------------------------------------------------------------------------- export function storePath(stateDir: string): string { return join(stateDir, "queue.json"); } export function loadStore(stateDir: string): QueueStore | null { try { const p = storePath(stateDir); if (!existsSync(p)) return null; const raw = JSON.parse(readFileSync(p, "utf8")) as QueueStore; if (raw && typeof raw === "object" && raw.items && typeof raw.items === "object") { // Backfill schema drift: older stores lack reviewerRunId/attempts (and // the legacy runId may be a bare short token). Normalize on read so // downstream code can rely on the fields existing. // MIGRATION: the pre-two-stage-review store used status "reviewing" for // the AI-review stage (renamed to "ai-review"; "human-review" is new). // Rewrite in memory so reads treat them as the current stage; persisted // on the next save. for (const it of Object.values(raw.items)) { // Legacy value not in the current QueueStatus union: widen to string so // the comparison is a deliberate migration check, not a typo. if ((it.status as string) === "reviewing") it.status = "ai-review"; } // Pre-rev stores start at revision 0 (same backfill-on-read rule as the // item fields below) — the first mutateStore write stamps rev 1. if (typeof raw.rev !== "number" || !Number.isFinite(raw.rev)) raw.rev = 0; if (typeof raw.revBy !== "string") raw.revBy = null; for (const it of Object.values(raw.items)) { if (it.reviewerRunId === undefined) it.reviewerRunId = null; if (it.specPassRunId === undefined) it.specPassRunId = null; if (it.pendingSpecPass === undefined) it.pendingSpecPass = false; if (it.timeoutMs === undefined) it.timeoutMs = null; if (it.attempts === undefined) it.attempts = 0; if (it.failCause === undefined) it.failCause = null; if (it.recoveries === undefined) it.recoveries = 0; if (it.recoveryNotBefore === undefined) it.recoveryNotBefore = null; if (it.recoveryEscalated === undefined) it.recoveryEscalated = false; if (it.runId === undefined) it.runId = null; if (it.cwd === undefined) it.cwd = null; if (it.blocker === undefined) it.blocker = null; if (it.blockedFrom === undefined) it.blockedFrom = null; if (it.shippedAt === undefined) it.shippedAt = null; if (it.reviewNotes === undefined) it.reviewNotes = null; if (it.acceptanceEvidence === undefined) it.acceptanceEvidence = null; // The ready boolean was folded into the status: approved = dispatchable, // approved+!ready → blocked. Normalize old stores on read. if (it.status === "approved" && (it as { ready?: boolean }).ready === false) it.status = "blocked"; delete (it as { ready?: boolean }).ready; } return raw; } return null; } catch { return null; } } /** RAW whole-file write (atomic tmp+rename, so a reader never sees a torn * file). It does NOT serialize against other writers and does NOT bump `rev` * — a bare loadStore → mutate → saveStore pair is exactly the lost-update * race this module now guards against. Every mutation goes through * mutateStore — which is the only non-test caller left (ensureMigrated writes * through it too). Exported for tests, which need to plant store states the * protocol would never produce. */ export function saveStore(stateDir: string, store: QueueStore): void { const p = storePath(stateDir); mkdirSync(stateDir, { recursive: true }); const tmp = `${p}.tmp-${process.pid}`; writeFileSync(tmp, JSON.stringify(store, null, 2) + "\n", "utf8"); renameSync(tmp, p); } export function newStore(): QueueStore { return { version: 1, rev: 0, revBy: null, items: {} }; } /** The store-ownership helper both hosts used to inline (loadStore ?? newStore). * One place, in the store layer. */ export function loadStoreOrNew(stateDir: string): QueueStore { return loadStore(stateDir) ?? newStore(); } // --------------------------------------------------------------------------- // The ONE safe mutation path (lock + compare-and-swap) // --------------------------------------------------------------------------- // // saveStore is atomic per write, but every caller used to do // loadStore → mutate → saveStore with no serialization: a writer that loaded // before another's rename silently erased it, whole-file (observed live — a // fully-specified proposal vanished with a success receipt, and its key was // handed out twice). The writers span PROCESSES (pi host, opencode host, // tools) and harness timers, so an in-process mutex is not enough. // // Two mechanisms, layered, and BOTH carry the writer's identity: // 1. an advisory LOCK FILE around the read-modify-write window — // cross-process, stamped with the holder's nonce, and published // ATOMICALLY (write a private tmp file, then link it into place) so a // contender never observes a half-created lock; // 2. an identified COMPARE-AND-SWAP as the correctness floor — the mutation // refuses to write over a revision newer than the one it read, and treats // its write as landed only when the re-read shows ITS OWN nonce at // rev+1 while the lock is still its own. // Identity is the whole point. A bare `rev === baseRev + 1` check is satisfied // by a COMPETITOR's stamp of the same number: two writers that both read // revision N both write N+1, both re-read N+1, and both report success while // one of the two writes is gone — bit for bit the incident this module exists // to prevent. So the lock is stamped with a nonce and the store records the // nonce that produced its revision; a writer only ever recognises its own. // The lock can still be stolen (a hard-killed holder must not wedge the // harness, so a stale/expired lock IS broken open) — the identity checks are // what make that steal safe FOR THE WRITER WHOSE LOCK WAS TAKEN: it sees that // it no longer holds it, counts a conflict, and re-applies on the fresh store. // The guarantee is ASYMMETRIC, and the residual is the THIEF's: the victim's // pre-write guard can pass a moment before the steal, so its rename is already // in flight and can land ON TOP of a write the thief has already made, verified // and reported. The victim then re-applies (its own change is never lost), but // the thief's receipt was for a write that is gone. That needs a steal (5s of // contention, or a stale/dead-pid lock) PLUS that interleaving, where before // the lock+CAS a loss needed nothing but two overlapping writers — which is why // this shipped with the hole named rather than hidden. Closing it fully needs a // different commit point: an O_EXCL rev-marker rename protocol on the store // file, where the marker create (not the store rename) decides the winner, so // no writer can be overwritten after it has been told it won. Deferred. A // cheaper partial mitigation was evaluated and also deferred: re-probing lock // ownership between saveStore's tmp write and its rename would shrink the // victim's exposure from a whole serialize+write to a single syscall, but it is // a real behaviour change (a new abort path, and saveStore would have to take a // nonce), so it belongs to the rev-marker work, not here. // Liveness is the hard requirement: NOTHING here waits unboundedly — a stuck // sweep would be a worse failure than the bug. const LOCK_STALE_MS = 15_000; const LOCK_WAIT_MS = 5_000; /** How long an UNREADABLE lock file is given before it counts as abandoned. * Unparsable content is not proof of abandonment — it is also what a lock * looks like for the instant between its creation and its content landing on * a filesystem that reorders them. Breaking one open immediately would let a * contender delete a lock a live writer had just taken. */ const LOCK_UNREADABLE_GRACE_MS = 250; const MAX_MUTATE_ATTEMPTS = 8; const MAX_LOCK_ROUNDS = 500; interface LockInfo { pid: number; host: string; at: number; /** the holder's WRITE IDENTITY — the same nonce it stamps into store.revBy. * "" for a lock written by a pre-nonce build (never matches a live nonce, * so such a lock is simply never mistaken for ours). */ nonce: string; } /** A globally unique write identity. Two writers can stamp the same `rev`; * they can never stamp the same nonce. */ function newWriteNonce(): string { return `${process.pid}@${hostname()}:${randomBytes(8).toString("hex")}`; } function lockPath(stateDir: string): string { return `${storePath(stateDir)}.lock`; } /** Block the thread for ~ms. The store writers are synchronous end to end, so * the wait has to be too; holds are file-IO short. */ function sleepSync(ms: number): void { if (ms <= 0) return; Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms); } function readLock(p: string): LockInfo | null { try { const raw = JSON.parse(readFileSync(p, "utf8")) as LockInfo; if (!raw || typeof raw.pid !== "number" || typeof raw.at !== "number") return null; return { ...raw, nonce: typeof raw.nonce === "string" ? raw.nonce : "" }; } catch { return null; } } /** Do WE hold the lock right now? The steal-safety check: any other writer * entering the critical section must first replace this file with its own, so * a lock that is missing or carries someone else's nonce means our window was * taken and whatever we are about to write (or just wrote) is unsafe. */ function holdsStoreLock(stateDir: string, nonce: string): boolean { const info = readLock(lockPath(stateDir)); return info !== null && info.nonce === nonce; } /** Is the lock holder gone (so the lock may be broken open)? Older than the * stale timeout = abandoned; a dead pid on THIS host = abandoned (the pid * check is only meaningful on the machine that wrote it); unreadable = only * once it has sat unreadable for the grace window (see the constant). */ function lockHolderGone(p: string, now: number, staleMs: number): boolean { const info = readLock(p); if (!info) { let mtimeMs: number; try { mtimeMs = statSync(p).mtimeMs; } catch { return true; // already gone from under us — nothing to wait for } return now - mtimeMs > LOCK_UNREADABLE_GRACE_MS; } if (now - info.at > staleMs) return true; if (info.host === hostname() && info.pid !== process.pid) { try { process.kill(info.pid, 0); } catch (e) { if ((e as { code?: string }).code === "ESRCH") return true; } } return false; } function acquireStoreLock(stateDir: string, nonce: string, waitMs: number, staleMs: number): void { const p = lockPath(stateDir); mkdirSync(stateDir, { recursive: true }); const deadline = Date.now() + waitMs; // The lock is published CONTENT-FIRST: fill a private tmp file, then link it // into place. link() is the exclusive-create primitive here, so the lock // never exists in the empty/half-written state that an open("wx") + // write-afterwards sequence leaves visible — a contender that read the lock // in that gap saw unparsable content, called it abandoned, and deleted a lock // its live owner was still about to write into (two holders, no dead pid, no // staleness, no expired wait budget). // CONSTRAINT: this requires a filesystem with HARDLINKS. A link() that fails // with anything but EEXIST (EPERM/ENOSYS/EOPNOTSUPP on some network/FUSE // mounts) propagates and fails the write, loudly. That is deliberate: the // obvious fallback — openSync(p, "wx") then write the content — is exactly // the half-created-lock hazard above, so falling back to it would trade a // clear failure for the silent two-holders bug. The state dir is a local // ~/.local/state path, so this does not arise; if it ever must live on a // hardlink-less mount, the fix is a directory lock (mkdir is atomic-exclusive // everywhere), not a wx fallback. const tmp = `${p}.tmp-${process.pid}-${randomBytes(6).toString("hex")}`; try { for (let round = 0; round < MAX_LOCK_ROUNDS; round++) { try { writeFileSync(tmp, JSON.stringify({ pid: process.pid, host: hostname(), at: Date.now(), nonce } satisfies LockInfo)); linkSync(tmp, p); return; } catch (e) { if ((e as { code?: string }).code !== "EEXIST") throw e; } const now = Date.now(); if (lockHolderGone(p, now, staleMs) || now >= deadline) { // NEVER WEDGE THE HARNESS: a dead/abandoned holder, or one that // outlived the wait budget, gets its lock broken open rather than // waited on. The identified CAS below is what keeps a broken-open lock // from losing a write — the writer we stole from will see that the lock // is no longer its own and re-apply instead of reporting success. try { unlinkSync(p); } catch { // someone else broke it first — just retry the create } continue; } sleepSync(4 + Math.floor(Math.random() * 12)); } } finally { try { unlinkSync(tmp); } catch { // the tmp link is scratch — the lock file survives it } } throw new Error(`queue: could not acquire the store lock at ${p} (contended past ${MAX_LOCK_ROUNDS} rounds)`); } function releaseStoreLock(stateDir: string, nonce: string): void { // Only unlink OUR lock: if it was broken open and re-taken while we ran, the // file belongs to the new holder and deleting it would strand them. if (!holdsStoreLock(stateDir, nonce)) return; try { unlinkSync(lockPath(stateDir)); } catch { // best-effort — a missing lock is already released } } /** The on-disk revision AND the identity that stamped it — together, the CAS * comparand. (0/null when the store is absent or unreadable.) */ function diskStamp(stateDir: string): { rev: number; revBy: string | null } { const store = loadStore(stateDir); return { rev: store?.rev ?? 0, revBy: store?.revBy ?? null }; } export interface MutateOptions { /** bounded CAS retries before the mutation ABORTS loudly (default 8) */ maxAttempts?: number; /** how long to wait for a contended lock before breaking it open (default 5s) */ lockWaitMs?: number; /** a lock older than this is abandoned (default 15s) */ lockStaleMs?: number; } /** * THE safe mutation path: load → apply → write, serialized across processes and * verified by compare-and-swap. `apply` receives the CURRENT store and mutates * it in place (returning whatever the caller needs to report); it may be re-run * on a conflict, so it must be a pure function of the store it is handed — no * spawns, no git, no nested mutateStore, no side effects that must happen once. * Do those before or after the call. * * Throws when the mutation could not be applied: an ABORT is the contract, so a * caller can never report success for a write that did not survive. An * exception from `apply` itself (an illegal transition, say) propagates * unchanged and writes nothing. */ export function mutateStore(stateDir: string, apply: (store: QueueStore) => T, opts: MutateOptions = {}): T { const maxAttempts = Math.max(1, opts.maxAttempts ?? MAX_MUTATE_ATTEMPTS); const lockWaitMs = opts.lockWaitMs ?? LOCK_WAIT_MS; const lockStaleMs = opts.lockStaleMs ?? LOCK_STALE_MS; let conflicts = 0; for (let attempt = 1; attempt <= maxAttempts; attempt++) { // A FRESH identity per attempt: it stamps the lock we take and the revision // we write, so both the "is this still my window?" and the "is this my // write?" questions have an answer no competitor can accidentally give. const nonce = newWriteNonce(); acquireStoreLock(stateDir, nonce, lockWaitMs, lockStaleMs); let landed: { value: T } | null = null; try { const store = loadStoreOrNew(stateDir); const baseRev = store.rev ?? 0; const value = apply(store); if (diskStamp(stateDir).rev !== baseRev || !holdsStoreLock(stateDir, nonce)) { // Our window was taken (the lock is someone else's now) or the store // moved under us — either way, re-apply on the fresh store. // ORDER IS DELIBERATE: the rev term reads and parses the WHOLE store // (milliseconds at real store sizes), so it runs FIRST; the cheap lock // probe runs last, leaving only a lock read between the check and the // write. Both terms are pure reads, so this is evaluation order only — // it narrows the check-then-act gap at zero cost. It does not close it // (see the residual above). conflicts++; } else { store.rev = baseRev + 1; store.revBy = nonce; saveStore(stateDir, store); const after = diskStamp(stateDir); // OUR nonce at OUR revision, with the window still ours. A competitor's // rev+1 does not count: that is exactly the write that erased us. if (after.rev === baseRev + 1 && after.revBy === nonce && holdsStoreLock(stateDir, nonce)) landed = { value }; else conflicts++; // someone wrote over us — our change did NOT survive } } finally { releaseStoreLock(stateDir, nonce); } if (landed) return landed.value; sleepSync(3 + Math.floor(Math.random() * 12 * attempt)); // backoff + jitter } throw new Error( `queue: store mutation ABORTED after ${maxAttempts} attempts — ${conflicts} concurrent write conflict(s) on ${storePath(stateDir)}; the change was NOT saved`, ); } /** One-time migration: import a legacy state.md into the programmatic store. * Host-agnostic — this is STORE logic (pi used to inline it; opencode never * got it). Best-effort: a broken migration must not break activation, and an * absent store is a no-op. */ export function ensureMigrated(stateDir: string): void { try { if (loadStore(stateDir)) return; const mdPath = join(stateDir, "state.md"); if (!existsSync(mdPath)) return; const migrated = migrateFromMd(readFileSync(mdPath, "utf8")); // Through the safe path like every other writer: two hosts can activate at // once, and the loser must see the winner's store rather than overwrite it. mutateStore(stateDir, (store) => { if (Object.keys(store.items).length > 0) return; // another activation migrated first store.items = migrated.items; }); const archived = `${mdPath}.migrated-${Date.now()}`; try { renameSync(mdPath, archived); } catch { // keep the original if rename fails — the store is authoritative now } } catch { // migration is best-effort; autopilot still works on an empty store } } export function queueLengths(store: QueueStore): QueueLengths { const out: QueueLengths = { proposal: 0, approved: 0, blocked: 0, active: 0, "ai-review": 0, "human-review": 0, failed: 0, done: 0, rejected: 0, "no-change": 0 }; for (const it of Object.values(store.items)) { if (isValidStatus(it.status)) out[it.status]++; // corrupt statuses (missing/invalid) are not counted here — they surface // in queryItems (list all) so they can be repaired } return out; } export interface QueueQuery { /** filter by one or more statuses (omitted = all) */ status?: QueueStatus | QueueStatus[]; /** ISO timestamp — only items with updatedAt >= since (last change filter) */ since?: string; sort?: "updatedAt" | "createdAt" | "key"; // default updatedAt desc limit?: number; // default 50 /** include the heavy detail fields (scope/evidence/value/urgency/risk/notes) */ includeNotes?: boolean; } /** * Query the store: filter by status / last-change, sort, cap, and project a * compact view (heavy detail fields only on request — keeps the LLM's * context small). Deterministic — no parsing anywhere. */ export function queryItems(store: QueueStore, q: QueueQuery = {}): Array & { key: string }> { let items = Object.values(store.items); const statuses = q.status === undefined ? null : Array.isArray(q.status) ? q.status : [q.status]; if (statuses) items = items.filter((i) => statuses!.includes(i.status)); if (q.since) items = items.filter((i) => i.updatedAt >= q.since!); items = [...items]; switch (q.sort ?? "updatedAt") { case "key": items.sort((a, b) => a.key.localeCompare(b.key)); break; case "createdAt": items.sort((a, b) => a.createdAt.localeCompare(b.createdAt)); break; default: items.sort((a, b) => b.updatedAt.localeCompare(a.updatedAt)); } items = items.slice(0, q.limit ?? 50); return items.map((i) => { const base = { key: i.key, status: i.status, blocker: i.blocker, title: i.title, runId: i.runId, reviewerRunId: i.reviewerRunId, // the reviewer-in-flight fact — an ai-review item with a reviewerRunId is ALREADY dispatched (do NOT queue_review it) // Budget-governance read facts: the cap the item runs under, and WHY it // failed (budget-capped vs verdict) — the operator-facing 'panel' read // must distinguish a run that died at its budget from one that failed on // a verdict without loading notes. timeoutMs: i.timeoutMs, failCause: i.failCause ?? null, shippedAt: i.shippedAt ?? null, // The worker model override (AUTOPILOT-79); absent = inherit the session's. ...(i.model ? { model: i.model } : {}), // The capability level (AUTOPILOT-110); absent = no tier. ...(i.tier ? { tier: i.tier } : {}), // DEPENDENCY GATE (KEY: AUTOPILOT-81): `after` shown whenever declared; // `waiting` (the CURRENTLY unmet subset) only when non-empty — an item // whose deps all finished shows `after` with no `waiting`, which is the // signal it is simply ready like any other approved item. ...(i.after && i.after.length ? { after: i.after } : {}), ...(i.after && i.after.length && unmetDependencies(i, store).length ? { waiting: unmetDependencies(i, store) } : {}), updatedAt: i.updatedAt, }; if (q.includeNotes) { return { ...base, scope: i.scope, evidence: i.evidence, value: i.value, urgency: i.urgency, risk: i.risk, notes: i.notes }; } return base; }); } // --------------------------------------------------------------------------- // Series registry — cwd → id series, AUTO-MAINTAINED by the framework // --------------------------------------------------------------------------- /** The prefix is a PROPERTY OF THE WORKSTREAM, discovered from history and * remembered by the framework — the agent's only job is to not fight it. * Resolution order (resolveSeries): registry → history (dominant series among * items with the same cwd) → repo-name slug → default "Q". Every add with a * cwd RECORDS the used series, so judgment is needed exactly once per * workstream — the first item — and never again. */ export interface SeriesRegistryEntry { series: string; updatedAt: string; } function seriesRegistryPath(stateDir: string): string { return join(stateDir, "series-registry.json"); } export function readSeriesRegistry(stateDir: string): Record { try { const p = seriesRegistryPath(stateDir); if (!existsSync(p)) return {}; const raw = JSON.parse(readFileSync(p, "utf8")) as Record; return raw && typeof raw === "object" ? raw : {}; } catch { return {}; } } /** Record cwd → series (best-effort, never throws — a registry problem must * not fail an add; the next add simply re-derives). */ export function recordSeries(stateDir: string, cwd: string, series: string): void { try { const reg = readSeriesRegistry(stateDir); reg[cwd] = { series, updatedAt: new Date().toISOString() }; const p = seriesRegistryPath(stateDir); mkdirSync(stateDir, { recursive: true }); const tmp = `${p}.tmp-${process.pid}`; writeFileSync(tmp, JSON.stringify(reg, null, 2) + "\n", "utf8"); renameSync(tmp, p); } catch { // best-effort } } /** The slug fallback: repo basename, uppercased + sanitized. */ export function seriesSlugFor(cwd: string): string { const base = (cwd.split("/").filter(Boolean).pop() ?? "").toUpperCase().replace(/[^A-Z0-9_-]/g, ""); return base || "Q"; } /** Dominant series among existing items with the same cwd (history beats the * slug — it survives repo renames and preserves sub-series). Same counting * rule as nextKeyFor: the series is the text BEFORE the first digit * (`B48-EARLIER` → B, `EVAL-EXPT-M9` → EVAL-EXPT-M). Most frequent wins; * ties broken by most recently updated. `excludeKey` skips one item (the * caller's own key) so a freshly-specified provisional Q handle cannot vote * for its own provisional series. */ function historicalSeries(store: QueueStore, cwd: string, excludeKey?: string): string | null { const counts = new Map(); for (const it of Object.values(store.items)) { if (it.cwd !== cwd) continue; if (excludeKey && it.key === excludeKey) continue; const m = /^([A-Za-z0-9_-]*?)(?:\d|$)/.exec(it.key); let series = m?.[1].replace(/[-_]+$/, ""); if (!series) continue; // A DIGIT-LESS key votes only its FIRST segment as the series. The old // rule let the whole key-without-digits become its own series, so one // legacy key (e.g. "MBR-HANDOVER-SAEID", "AUTOPILOT-SCHEDULED-OFF") // latched as the workstream identity and every new item inherited the // weird prefix (observed twice: MBR-HANDOVER-SAEID-1, AUTOPILOT- // WORKTREE-PRESERVATION-1). if (!/\d/.test(it.key)) { series = (series.split(/[-_]/)[0] || series).trim(); if (!series) continue; } const cur = counts.get(series) ?? { n: 0, latest: 0 }; counts.set(series, { n: cur.n + 1, latest: Math.max(cur.latest, Date.parse(it.updatedAt) || 0) }); } let best: string | null = null; for (const [series, v] of counts) { if (best === null) { best = series; continue; } const bv = counts.get(best)!; if (v.n > bv.n || (v.n === bv.n && v.latest > bv.latest)) best = series; } return best; } /** Resolve the series for a cwd: registry → history → slug. Records nothing — * the caller records the series it actually used. `excludeKey` excludes one * item from the history vote (the item being renamed must not vote for its * own provisional series — it would win updatedAt ties). */ export function resolveSeries(stateDir: string, cwd: string, opts?: { excludeKey?: string }): string { if (!cwd) return "Q"; const hit = readSeriesRegistry(stateDir)[cwd]; if (hit?.series) return hit.series; const store = loadStore(stateDir); if (store) { const hist = historicalSeries(store, cwd, opts?.excludeKey); if (hist) return hist; } return seriesSlugFor(cwd); } export function itemByRunId(store: QueueStore, runId: string): QueueItem | null { for (const it of Object.values(store.items)) { if (it.runId && runIdMatches(runId, it.runId)) return it; } return null; } /** Find an `ai-review` item whose REVIEWER run matches (queue_review attribution). */ export function itemByReviewerRunId(store: QueueStore, runId: string): QueueItem | null { for (const it of Object.values(store.items)) { if (it.status === "ai-review" && it.reviewerRunId && runIdMatches(runId, it.reviewerRunId)) return it; } return null; } /** * Find the `ai-review` item a run's TASK names (the `KEY: ` line every * harness task starts with) — the FALLBACK attribution for a reviewer * completion whose dispatch-time `reviewerRunId` link is missing (KEY: * AUTOPILOT-104). Scoped to `ai-review` exactly like `itemByReviewerRunId`, * AND to items that carry NO reviewer ref: the missing-ref case is the * discriminator this fallback exists for, whereas a DIFFERENT ref is ambiguous * (another round's reviewer) and must be surfaced by the caller, never guessed * through. `null`/absent key → null (the backend could not read the task, or * no task named a key). */ export function itemByTaskKey(store: QueueStore, key: string | null | undefined): QueueItem | null { if (typeof key !== "string" || !key) return null; const it = store.items[key]; return it && it.status === "ai-review" && !it.reviewerRunId ? it : null; } /** Find a `proposal` item whose SPEC PASS run matches (KEY: AUTOPILOT-36). * Scoped to `proposal` the same way `itemByReviewerRunId` is scoped to * `ai-review`: leaving proposal clears `specPassRunId` (updateItem's * leaving-proposal stamping, above), so a completion for an item that was * approved/rejected/deferred while the pass ran already finds nothing here * — the same "item moved on, no orphaned write" guarantee AUTOPILOT-59's * claim-before-spawn gives human re-dispatch. */ export function itemBySpecPassRunId(store: QueueStore, runId: string): QueueItem | null { for (const it of Object.values(store.items)) { if (it.status === "proposal" && it.specPassRunId && runIdMatches(runId, it.specPassRunId)) return it; } return null; } export function runIdMatches(fullRunId: string, token: string): boolean { const a = fullRunId.toLowerCase(); const b = token.toLowerCase(); const n = Math.min(a.length, b.length, 8); if (n < 6) return false; return a.slice(0, n) === b.slice(0, n); } // --------------------------------------------------------------------------- // Dependency gate (KEY: AUTOPILOT-81) — `after` keys, validated at every // write, and the one predicate every dispatch lane shares for "is this item // blocked on unfinished work". // --------------------------------------------------------------------------- /** Keys in `item.after` whose current status is not yet `done` — the item is * WAITING on these. `done` is the release condition (not human-review, not * shippedAt — shipping can be a deliberate, later step). A key that no * longer resolves in the store still counts as unmet (should not happen — * `validateAfter` refuses an unknown key at write time — but an item can * never be ready on a dependency it cannot verify). Absent/empty `after` = * nothing unmet. * DELIBERATELY `done`-only — NOT the terminal set (AUTOPILOT-113): a * `rejected`/`no-change` dependency is CLOSED but produced no ACCEPTED work * product, so the dispatch gate stays shut until a human acts. Alongside the * `done`-only shipping/finisherSource branches, this intentionally does not read * `isTerminalStatus`; contrast the release sweep in updateItem (which DOES * release a park on any terminal dependency). Pinned by "a rejected dependency * stays unmet". */ export function unmetDependencies(item: Pick, store: QueueStore): string[] { if (!item.after || item.after.length === 0) return []; return item.after.filter((key) => store.items[key]?.status !== "done"); } /** Does setting `key`'s `after` to `after` introduce a cycle, given the * store's EXISTING `after` edges elsewhere? Walks forward from each proposed * dependency; a path back to `key` is a cycle. Assumes the store holds no * pre-existing cycle — an invariant this same validation maintains on every * write — so a per-walk `visited` set is enough to terminate on a corrupt * store too (it just may not find every cycle in one already there). */ function findAfterCycle(store: QueueStore, key: string, after: string[]): string[] | null { function walk(from: string, target: string, visited: Set): string[] | null { if (from === target) return [from]; if (visited.has(from)) return null; visited.add(from); for (const next of store.items[from]?.after ?? []) { const path = walk(next, target, visited); if (path) return [from, ...path]; } return null; } for (const dep of after) { const path = walk(dep, key, new Set()); if (path) return [key, ...path]; } return null; } /** Validate a proposed `after` list for `key` against the store: every key * must exist, no self-reference, no cycle. Returns a clear rejection reason * string, or null when valid. The ONE gate both `addItem` and `updateItem` * call — a bad `after` must never land silently (THE CHANGE #1). */ export function validateAfter(store: QueueStore, key: string, after: string[]): string | null { if (after.includes(key)) return `'after' cannot reference itself ('${key}')`; for (const dep of after) { if (!store.items[dep]) return `'after' references an unknown key '${dep}'`; } const cycle = findAfterCycle(store, key, after); if (cycle) return `'after' would create a dependency cycle: ${cycle.join(" \u2192 ")}`; return null; } /** The "Depends on" header prefixed to a dispatched task when `item.after` is * set (KEY: AUTOPILOT-81): tells the worker whether each dependency's work is * actually on main yet — `done` only means the WORK is done; shipping can be * a separate, later, manual step. Built from the dependency's recorded * `shippedAt`/`landedEvidence`, never by inspecting a repo (the dispatched * worker's own cwd may not even be the dependency's repo). Empty string when * `after` is unset/empty. */ export function dependsOnHeader(item: Pick, store: QueueStore): string { if (!item.after || item.after.length === 0) return ""; const lines = item.after.map((depKey) => { const dep = store.items[depKey]; if (!dep) return `Depends on: ${depKey} (unknown item — cannot verify)`; const sha = dep.landedEvidence?.sha ?? null; const shipped = dep.shippedAt ? (sha ? `yes @ ${sha.slice(0, 8)}` : `yes @ ${dep.shippedAt}`) : "no"; return `Depends on: ${depKey} (${dep.status}, shipped: ${shipped})`; }); return lines.join("\n"); } // --------------------------------------------------------------------------- // Mutations (validated) // --------------------------------------------------------------------------- export interface UpdatePatch { status?: QueueStatus; blocker?: BlockerReason; /** The status this park came from (see BlockedFrom) — recorded automatically * on a transition INTO `blocked`; an explicit value wins. */ blockedFrom?: BlockedFrom | null; runId?: string | null; /** See the QueueItem field doc — explicit here wins over updateItem's own * terminal-transition stamping (the SAME explicit-wins convention every * other auto-derived field in this patch follows). */ lastRunId?: string | null; reviewerRunId?: string | null; /** SPEC PASS run id (KEY: AUTOPILOT-36) — see the QueueItem field doc. */ specPassRunId?: string | null; /** SPEC PASS pending marker — see the QueueItem field doc. */ pendingSpecPass?: boolean; timeoutMs?: number | null; /** WORKER MODEL OVERRIDE — see the QueueItem field doc. */ model?: string; /** CAPABILITY LEVEL (KEY: AUTOPILOT-110) — see the QueueItem field doc. */ tier?: string; attempts?: number; failCause?: FailCause | null; recoveries?: number; recoveryNotBefore?: number | null; recoveryEscalated?: boolean; /** WHERE the current run's managed worktree lives (AUTOPILOT-47 C). */ runWorktree?: { runId: string; path: string; branch: string } | null; /** DEPENDENCY GATE — see the QueueItem field doc. Validated the same way * (`validateAfter`) before the write lands; an empty array clears it. */ after?: string[]; title?: string; scope?: string; cwd?: string | null; evidence?: string; value?: string; urgency?: string; risk?: string; notes?: string; /** SHIPPING MARKER (KEY: AUTO-SHIP-ON-DONE) — set by the merge-finisher * when it ships; cleared when the item leaves `done` (re-open). */ shippedAt?: string | null; /** FINISHER-EVIDENCE: the dispatch class, the dispatch-time HEAD baseline, * the landed proof, and the append-only override log. */ dispatchClass?: DispatchClass; finisherSource?: string | null; finisherBaseline?: FinisherBaseline | null; landedEvidence?: LandedEvidence | null; overrides?: FailureOverride[]; /** The reviewer's bounded caveats, captured at verdict parse (AUTOPILOT-27). */ reviewNotes?: ReviewNotes | null; /** HUMAN RE-DISPATCH pending marker (see QueueItem field doc). */ pendingRedispatchFindings?: string | null; /** WHERE to review this attempt, rebuilt per verdict PASS (AUTOPILOT-57). */ reviewTargets?: string[] | null; } /** Apply a validated update. Throws on an illegal transition. Returns the item. */ export function updateItem(store: QueueStore, key: string, patch: UpdatePatch, now = new Date().toISOString()): QueueItem { const item = store.items[key]; if (!item) throw new Error(`queue: no item '${key}'`); // Strip undefined fields BEFORE spreading — otherwise `{...patch}` would // clobber existing values (e.g. status) with undefined when the caller // omits a field. const clean: UpdatePatch = {}; for (const [k, v] of Object.entries(patch)) { if (v !== undefined) (clean as Record)[k] = v; } // DEPENDENCY GATE (KEY: AUTOPILOT-81): validate BEFORE anything else lands // — a bad `after` must reject the whole write, not land partially. if (clean.after !== undefined) { const afterErr = validateAfter(store, key, clean.after); if (afterErr) throw new Error(`queue: ${afterErr}`); } const from = item.status; const to = clean.status; // Leaving `failed` (re-dispatch to active, verified-complete to done, human // re-open to approved, drop to rejected) clears the failure cause — a stale // 'budget-capped' flag must not describe the NEXT run of the same item. // (A metadata-only patch with status undefined leaves the cause alone, as // does a re-statement of status failed while the item stays failed.) // EXCEPT a PARK (failed → blocked, KEY: AUTOPILOT-49): the item is held // BECAUSE of how it failed, so the cause is what the hold is about — and it // is the machine-readable marker that separates "parked after a dispatch" // from a proposal that was never dispatched (auto-recovery surfaces the // former once). It is cleared on the way OUT of blocked, below, so it never // rides into the released run. const parking = from === "failed" && to === "blocked"; if (to !== undefined && to !== "failed" && !parking) clean.failCause = null; // LEAVING `blocked` drops the hold markers unless the caller re-states them: // the blocker reason describes the wait, not the work after it, and an // approved item still labelled `decision` reads as blocked in every render. // `recoveryEscalated` is cleared for the same reason — it means "already // announced, and the recovery phases stay off this item", so a RELEASE that // left it set would hand back an item auto-recovery is permanently deaf to // (no retry, and no escalation tick the next time it fails). A deliberate // release re-arms the item; that is what makes the park releasable rather // than a one-way strand. if (from === "blocked" && to !== undefined && to !== "blocked") { if (clean.blocker === undefined) clean.blocker = null; if (clean.recoveryEscalated === undefined) clean.recoveryEscalated = false; // The park is over, so its recorded origin is no longer meaningful — a // later park re-records it. Keeping a stale origin would let a release // restore this item to a status it was never parked from. if (clean.blockedFrom === undefined) clean.blockedFrom = null; } // Leaving `failed` also drops the recovery SCHEDULING gate (the backoff is // per-failure-episode) — a manual re-dispatch or a recovery must never // inherit a stale "not before" instant. The attempt COUNTER + the escalation // flag deliberately persist (the recovery bound spans the item's history). if (to !== undefined && to !== "failed" && clean.recoveryNotBefore === undefined) clean.recoveryNotBefore = null; // done/human-review/failed/no-change are terminal-ish for their runs: the AI review // is over at human-review, so clear stale run refs (a done item must not // keep pointing at a dead worker; the FAIL flip already nulls them; this // covers the manual/orchestrator paths). Before nulling, stamp `lastRunId` // with the identity of the run being cleared (KEY: AUTOPILOT-85) — // resolveWorkTip's runWorktree gate (shipping.ts) needs SOME live runId to // compare against once `runId` itself is gone, or a done item's own attempt // reads as stale forever. The value being cleared is whatever runId is IN // EFFECT right before this line runs: an explicit patch.runId (already // folded into `clean`) wins over the item's stored one, matching the rest of // this function's explicit-wins convention. A null/absent effective runId // (already-terminal item re-stated, or a repair with no run ever attached) // leaves the existing `lastRunId` alone — there is nothing new to attribute // the terminal state to. if (to === "done" || to === "human-review" || to === "failed" || to === "no-change") { const clearedRunId = clean.runId !== undefined ? clean.runId : item.runId; if (clean.lastRunId === undefined && clearedRunId) clean.lastRunId = clearedRunId; clean.runId = null; clean.reviewerRunId = null; } // ENTERING ai-review starts a NEW review round, so the previous round's // reviewer ref must not survive into it (a stale id blocks queue_review and // silently stops the harness's auto-review). Only on an ACTUAL entry: an // ai-review → ai-review re-statement (a metadata patch that repeats the // status) must NOT drop the reviewer that is running right now. An explicit // reviewerRunId in the same patch wins — the caller is stamping the dispatch. if (to === "ai-review" && from !== "ai-review" && clean.reviewerRunId === undefined) clean.reviewerRunId = null; // LEAVING `proposal` (approve/reject/defer) drops a spec pass in flight // (KEY: AUTOPILOT-36): the pass is ONLY meaningful while the item is still // a decidable proposal — once the human has approved/rejected/deferred it, // a late spec-pass completion has nothing to write back to (core.ts's // completion handler re-checks status "proposal" anyway, but clearing the // ref here means a stray completion for a superseded run is never even // attributed). Only on an ACTUAL exit — a metadata-only patch that leaves // `to` undefined (e.g. refine) must not disturb a pass genuinely in flight. if (to !== undefined && to !== "proposal" && clean.specPassRunId === undefined) clean.specPassRunId = null; if (to !== undefined && to !== "proposal" && clean.pendingSpecPass === undefined) clean.pendingSpecPass = false; // ENTERING `active` starts a NEW run, so the PREVIOUS run's landed evidence // AND its dispatch baseline must not survive into it (FINISHER-EVIDENCE): a // re-dispatched finisher that genuinely does nothing has to fail on its own // merits, and either leftover would launder that failure into a landing (a // stale baseline makes the PREVIOUS run's commit look like this run's work). // Both are cleared HERE, on the lifecycle edge, so every lane that returns an // item to `active` is covered — not just the dispatch tool. A lane that // spawns a finisher passes a FRESH baseline in the same patch (explicit wins); // the override LOG (overrides[]) and finisherSource keep the history. if (to === "active" && from !== "active" && clean.landedEvidence === undefined) clean.landedEvidence = null; if (to === "active" && from !== "active" && clean.finisherBaseline === undefined) clean.finisherBaseline = null; // ENTERING `active` from anywhere OTHER than the human-redispatch move // (panels.ts OR queue_update's `findings` param sets pendingRedispatchFindings // explicitly IN THE SAME patch — see the field's doc comment) must not // inherit a stale pending marker from an earlier, already-consumed human // redispatch — a dead marker would re-arm dispatchPendingHumanRedispatch // against this run's own scope with someone else's old findings. if (to === "active" && from !== "active" && clean.pendingRedispatchFindings === undefined) clean.pendingRedispatchFindings = null; // blocked must say WHY (parked/serialized/merge/decision) — a blocker-less // block would be indistinguishable from a rejected proposal. if (to === "blocked" && !clean.blocker) { throw new Error(`queue: blocked requires a blocker reason (parked/serialized/merge/decision) for '${key}'`); } // A `serialized` park NAMES what it waits for (KEY: AUTOPILOT-109). A // serialized park with no `after` is a CONTRADICTION: `serialized` means // "waiting on a named dependency", so with no dependency there is nothing // that can ever satisfy it — the one combination that can never self-release // used to be the easiest to create. It is REFUSED at write time instead of // stored. An INDEFINITE human deferral is not `serialized`: the correct // reason is `parked`/`decision`, a different hold with a different (human) // release. The EFFECTIVE state is checked (not just an explicit status // write), so REMOVING the `after` from an already-serialized item is refused // too — the contradiction must not be reachable by editing either side. const resultingStatus = to ?? from; const resultingBlocker = clean.blocker !== undefined ? clean.blocker : item.blocker; const resultingAfter = clean.after !== undefined ? clean.after : item.after; if (resultingStatus === "blocked" && resultingBlocker === "serialized" && (!resultingAfter || resultingAfter.length === 0)) { throw new Error( `queue: blocked+serialized requires a non-empty 'after' (nothing can release a serialized park with no dependency) for '${key}' — ` + `set 'after' to the key(s) it waits on, or use blocker 'parked'/'decision' for an indefinite human deferral`, ); } // A park records WHERE it came from (KEY: AUTOPILOT-109): the release sweep // restores `blockedFrom`, so it must be known. Only on an ACTUAL entry to // blocked (a re-statement of an already-blocked item must not overwrite the // origin), and only from a status that can be a park origin — the transition // table permits proposal/approved/failed (and human-review), and the repair // path for a corrupt status must not record garbage. if ( to === "blocked" && from !== "blocked" && clean.blockedFrom === undefined && (from === "approved" || from === "proposal" || from === "failed" || from === "human-review") ) { clean.blockedFrom = from; } // BLOCKED → AI-REVIEW (KEY: AUTOPILOT-93) is EVIDENCE-GATED: the table allows // the edge, but only an item that actually carries a completed deliverable may // take it (see completedDeliverableEvidence). The evidence is read from the // item's own record AND from THIS patch (explicit-wins), so the tool route can // record the completed run in the very write that moves it to review. An item // with no completed run is REFUSED here — never silently marked reviewable. if (from === "blocked" && to === "ai-review" && !completedDeliverableEvidence({ ...item, ...clean })) { throw new Error( `queue: illegal transition blocked → ai-review for '${key}': no evidence of a completed deliverable ` + `(need a completed run — runId/lastRunId — or a recorded branch/sha — runWorktree/landedEvidence/shippedAt). ` + `Record the completed run with queue_update({key:'${key}', completedRun:''}) then retry, or release via blocked → approved and dispatch.`, ); } // SHIPPING MARKER lifecycle: the marker is only meaningful while the item is // `done`. ANY transition OUT of done (re-open to approved, re-dispatch to // active, drop to rejected) clears it — a re-approved item starts a fresh // approval → shipping cycle and must ship again. Re-stating done (or a // metadata-only patch) leaves it alone. if (to !== undefined && to !== "done") clean.shippedAt = null; if (to !== undefined && to !== from && !validTransition(from, to)) { // allow REPAIR of a corrupt item (status missing/invalid) to any valid status if (!isValidStatus(from) && isValidStatus(to)) { // repair path — fall through } else { throw new Error(`queue: illegal transition ${from} → ${to} for '${key}'`); } } const next: QueueItem = { ...item, ...clean, updatedAt: now }; // A human re-open to `approved` starts a FRESH agent review loop — the // attempts counter is the agent-review FAIL cap, not the human's judgment. // BOTH terminal re-opens reset it (AUTOPILOT-113): `done → approved` (the // user found issues after approval) and `no-change → approved` (a deliberate // re-open of a closed refusal) — a prior life's spent FAILs must not count // against the fresh loop. `rejected` has no exit, so it never arrives here. if ((from === "done" || from === "no-change") && to === "approved") next.attempts = 0; store.items[key] = next; // DEPENDENCY RELEASE SWEEP (KEY: AUTOPILOT-109, generalizing AUTOPILOT-81). // The release used to fire ONLY when a dependency KEY reached `done`, and it // always restored `approved`. Both were wrong: a `rejected` dependency is // just as terminal, and an unconditional `approved` could promote a park that // was never approved. On ANY status write, every parked-on-a-dependency item // is re-evaluated: if it NAMES dependencies (non-empty `after`) and they are // ALL satisfied — TERMINAL (`isTerminalStatus`: `done`, `rejected`, or a // closed `no-change` refusal) — it is restored to the status it was PARKED // FROM (`blockedFrom`), with a note naming the dependency that completed. // Termination, not `done` alone, is the release condition (AUTOPILOT-113): a // `no-change` dependency delivered nothing but is CLOSED, so the dependent // must stop WAITING on it — while `unmetDependencies` stays `done`-only, so // the released dependent still cannot auto-dispatch on a dependency that // never delivered. A park from `proposal` returns to `proposal` (the human approval // gate is never skipped); `failed`/`human-review` return to their own lane. // NEVER auto-release: an item with no named dependency (a `parked`/`decision` // human gate, or legacy residue — see serializedWithoutDependency), a human // gate even if it happens to name dependencies, or an item with no recorded // origin (the release cannot know whether `approved` would skip a gate, so it // is left for a human). Gated on an actual status write so a metadata-only // patch never re-releases/re-notes. if (to !== undefined) { for (const dependent of Object.values(store.items)) { if (dependent.status !== "blocked") continue; // Nothing to satisfy: never auto-release a park with no named dependency // (the legacy residue, or a human gate with no `after`). if (!dependent.after || dependent.after.length === 0) continue; // Human gates are released by a human, never by a dependency completing. if (dependent.blocker === "parked" || dependent.blocker === "decision") continue; const restore = dependent.blockedFrom; if (restore !== "approved" && restore !== "proposal" && restore !== "failed" && restore !== "human-review") continue; const satisfied = dependent.after.filter((dep) => { const status = store.items[dep]?.status; return status !== undefined && isTerminalStatus(status); }); if (satisfied.length !== dependent.after.length) continue; // some dependency still pending // Name the dependency that completed: this write's own key when it is one // of them (the familiar "released: last dependency done" shape the // audit trail knows), else the last resolved one. const completed = satisfied.includes(key) ? key : satisfied[satisfied.length - 1]; // the completed dependency is TERMINAL, so the verb is its own status // (`done`/`rejected`/`no-change`) — never assume `done` (AUTOPILOT-113) const verb = store.items[completed]?.status ?? "done"; const note = `released: last dependency ${completed} ${verb}`; store.items[dependent.key] = { ...dependent, status: restore, blocker: null, blockedFrom: null, recoveryEscalated: false, // A released park that is NOT returning to `failed` must not carry a // stale failure cause into its next life (mirrors the leaving-blocked // rule); a return to `failed` keeps the cause its own lane retries on. ...(restore !== "failed" ? { failCause: null } : {}), notes: dependent.notes ? `${dependent.notes}\n\n${note}` : note, updatedAt: now, }; } } return next; } export function addItem( store: QueueStore, item: Omit, now = new Date().toISOString(), ): QueueItem { // DEPENDENCY GATE (KEY: AUTOPILOT-81) — same validation updateItem applies. if (item.after && item.after.length) { const afterErr = validateAfter(store, item.key, item.after); if (afterErr) throw new Error(`queue: ${afterErr}`); } // A directly-created `blocked` item records where it was parked from // (KEY: AUTOPILOT-109) so the release sweep can restore it. A caller that // predates the field (tests, migration) gets the historical default — // `approved` — which is exactly where the old release sent a // serialized dependent. Items created in any other status carry null. const full: QueueItem = { ...item, blockedFrom: item.blockedFrom ?? (item.status === "blocked" ? "approved" : null), createdAt: now, updatedAt: now, }; store.items[item.key] = full; return full; } // --------------------------------------------------------------------------- // Migration (one-time: import an existing state.md into the store) // --------------------------------------------------------------------------- export function migrateFromMd(md: string): QueueStore { const store = newStore(); // Tolerant: reuses the section+grouping logic from queue.ts via a lightweight // local scan — a key line starts an entry; continuation lines append. const lines = md.split("\n"); const sections = splitSections(lines); const statusFor = (name: string): QueueStatus | null => { if (name.startsWith("Active")) return "active"; if (name.startsWith("Approved")) return "approved"; if (name.startsWith("Backlog")) return "proposal"; if (name.startsWith("Reviewing")) return "ai-review"; if (name.startsWith("Completed")) return "done"; if (name.startsWith("Failed")) return "failed"; return null; }; for (const sec of sections) { const status = statusFor(sec.name); if (!status) continue; for (const { key, text } of groupEntries(lines, sec.start + 1, sec.end)) { const joined = text.join(" "); const runM = joined.match(/(?:worker|Worker)\s+(?:run\s+)?([0-9a-f]{6,})/); const statusM = joined.match(/status:\s*([A-Za-z]+)/); const title = text[0].slice(text[0].indexOf(":") + 1).trim(); const nonReady = /\b(BLOCKED|DISPATCHED|REMOVED|PARKED|SERIALIZED|HOLD|DEFERRED)\b/i.test(joined); const blocker = /\bPARKED\b/i.test(joined) ? "parked" : /\bSERIALIZED\b/i.test(joined) ? "serialized" : /\bBLOCKED\b/i.test(joined) ? "merge" : null; // The fold: approved + a non-ready marker in the text → blocked. const foldedStatus: QueueStatus = status === "approved" && nonReady ? "blocked" : status; addItem(store, { key, status: foldedStatus, blocker, title, // preserve the FULL original entry text (free-form) — nothing lost in the render scope: joined, cwd: null, evidence: "", value: "", urgency: "", risk: "", runId: runM ? runM[1] : null, reviewerRunId: null, specPassRunId: null, timeoutMs: null, attempts: 0, notes: joined, }); } } return store; } // --------------------------------------------------------------------------- // Internal helpers (kept local; queue.ts exports its own equivalents) // --------------------------------------------------------------------------- function splitSections(lines: string[]): Array<{ name: string; start: number; end: number }> { const out: Array<{ name: string; start: number; end: number }> = []; let cur: { name: string; start: number; end: number } | null = null; for (let i = 0; i < lines.length; i++) { const m = lines[i].match(/^##\s+([A-Za-z ]+)/); if (m) { if (cur) cur.end = i; cur = { name: m[1].trim(), start: i, end: lines.length }; out.push(cur); } } return out; } function groupEntries(lines: string[], start: number, end: number): Array<{ key: string; text: string[] }> { const entries: Array<{ key: string; text: string[] }> = []; let cur: { key: string; text: string[] } | null = null; for (let i = start; i < end; i++) { const line = lines[i]; if (!line.trim() || /^#{1,3}\s+/.test(line)) { cur = null; continue; } const key = entryKey(line); if (key) { cur = { key, text: [line] }; entries.push(cur); } else if (cur) { cur.text.push(line); } } return entries; } function entryKey(line: string): string | null { const m = line.match(/^([A-Z][A-Z0-9]*(?:-[A-Z0-9]+)*)(?:\s*\[[^\]]*\])?:\s/); return m ? m[1] : null; }