// core.ts — the orchestrator-autopilot decision engine. Framework-agnostic: // consumes subagent lifecycle events, owns a run ledger, derives the domain // signals (slot-freed / queue-low / capacity-gap) from the PROGRAMMATIC QUEUE // STORE (queue.json — extension-owned, machine-readable) and decides whether // to emit a tick. Adapters (pi/opencode/claude) wire this to their platform // events, wake channels, and queue tools. // // Design principle: the plugin guarantees the TRIGGER, the orchestrator keeps // the JUDGMENT. This engine never picks a task, never dispatches, never // approves. It ensures the orchestrator runs its loop at the right moments. // // Facts split: FLEET (running workers) is event-derived via the run ledger; // QUEUE (approved/ready items) comes from the queue store — both deterministic. // The old state.md parse survives ONLY as a transition-period fallback (the // store is migrated once from state.md, then md is retired). // // Deferred delivery (AUTOPILOT-6): a tick whose delivery the busy-deferral // held is RE-DERIVED from the live store at delivery (refreshTickFacts) — the // facts in the message are the facts at delivery, never a stale generation-time // snapshot ("X free, N ready" can never be shown while those items run). // // Fail-safe to action: when neither store nor md is available, we tick anyway // rather than risk a missed refill — a spurious tick costs one LLM turn, a // missed one costs an idle slot. import { readFileSync, writeFileSync, existsSync, renameSync, mkdirSync, appendFileSync } from "node:fs"; import { join } from "node:path"; import { loadStore, mutateStore, itemByRunId, itemByReviewerRunId, itemByTaskKey, itemBySpecPassRunId, updateItem, unmetDependencies, type QueueStore, type FailCause, } from "./queue-store.ts"; import type { Tick, TickReason, CompletionEvent, DomainEvent, AutopilotResult } from "./types.ts"; import type { AutopilotConfig } from "./config.ts"; import { parseVerdict, parseSpecPassScope, verdictTail, verdictBody } from "./verdict.ts"; import { extractReviewNotes } from "./review-notes.ts"; import { parseAcceptanceEvidence, storedAcceptanceEvidence } from "./acceptance-evidence.ts"; import { validateReviewReport } from "./review-contract.ts"; import { appendOverride, finisherLandedEvidence, isFinisherItem, landedNote, landedOverride } from "./finisher-evidence.ts"; import { collectRunIds } from "./run-ids.ts"; import { formatDurationMs } from "./duration.ts"; import { liveTrackedSlots, type OccupiedSlotsOpts } from "./framework/auto-dispatch.ts"; import type { SubagentBackend } from "./backends/types.ts"; // The snapshot shape the tick engine consumes (derived from the queue store). // The legacy md parser (queue.ts) is RETIRED — state.md exists only as a // one-time migration input, parsed by queue-store.migrateFromMd. interface QueueState { active: Array<{ key: string; runId?: string; status?: string; title: string; line: string; lineIndex: number; section: string; }>; approved: Array<{ key: string; title: string; line: string; lineIndex: number }>; // READY (dispatchable now) — unmet-dependency items are excluded (see `waiting`) /** DEPENDENCY GATE (KEY: AUTOPILOT-81): `approved` items whose `after` still * names an unfinished dependency — not ready despite being approved. */ waiting: Array<{ key: string; title: string; after: string[] }>; occupied: number; ready: number; /** The queue-known inventory (max of the event ledger + store-derived), * pre-fleet clamp — used for the fleet-status-vs-inventory parity note. */ tracked?: number; /** Items awaiting the user's approval decision (status: proposal). */ proposalsPending: number; ok: boolean; /** Budget telemetry: ACTIVE items with a recorded wall-clock budget * (timeoutMs) — remaining/cap per run, so dispatch ticks can surface cap * risk (~75% used) before the run is cut off. Derived from the item's * updatedAt (the flip-to-active dispatch stamp ≈ run start). */ budget: Array<{ key: string; cap: number; remaining: number; elapsed: number; used: number }>; /** Keys of FAILED items whose cause is budget-capped — the at-cap recovery * queue (re-dispatch with a bigger budget). */ budgetCappedFailed: string[]; } interface LedgerEntry { agent?: string; startedAt: number; } // --------------------------------------------------------------------------- // Autopilot // --------------------------------------------------------------------------- export class Autopilot { private cfg: Required> & AutopilotConfig; private ledger = new Map(); private lastTickAt = 0; private lastTickHash = ""; private lastSettledHash = ""; private lastReviewTickAt = 0; /** Runs already warned about approaching their budget (one warning per run * per Autopilot lifetime — the timer must not nag every sweep). */ private budgetWarned = new Set(); constructor(config: AutopilotConfig) { this.cfg = { maxSlots: 3, queueLowThreshold: 2, workerAgents: ["worker"], reviewerAgents: ["orchestrator-reviewer"], reviewCap: 5, quietPeriodMs: 60_000, zombieGraceMinutes: 30, ...config, }; } // -- public API ----------------------------------------------------------- /** The queue-store directory this engine reads. The runner's delivery-time * premise re-check (AUTOPILOT-87) shares it, so the router validates against * the SAME store the engine acts on. */ get stateDir(): string { return this.cfg.stateDir; } /** subagent:async-started → record the run in the ledger (fleet truth). */ handleAsyncStarted(runId: string, agent?: string | null, now = this.now()): AutopilotResult { if (!runId) return empty(); this.ledger.set(runId, { agent: agent ?? undefined, startedAt: now }); this.logEvent("started", { runId, agent }); return empty(); } /** * subagent:async-complete → attribute the run to the store item and flip * active→ai-review/failed. Does NOT tick — the adapter fetches the * AUTHORITATIVE fleet count (pi-subagents fleet status — the same source the * orchestrator's `subagent status` reads) and then sweeps, so the tick's * FLEET number can never contradict the orchestrator's own view. */ handleAsyncComplete(ev: CompletionEvent, now = this.now()): AutopilotResult { const topRunId = typeof ev.runId === "string" && ev.runId ? ev.runId : typeof ev.id === "string" ? ev.id : undefined; if (!topRunId) return empty(); const entry = this.ledger.get(topRunId); this.ledger.delete(topRunId); this.logEvent("complete", { runId: topRunId, agent: ev.agent, success: ev.success, status: ev.status }); const candidates = collectRunIds(ev); let outcome: "failed" | "ai-review" = ev.success === false || ev.timedOut ? "failed" : "ai-review"; let flipped = false; let flippedKey = ""; let failCause: FailCause | null = null; /** FINISHER-EVIDENCE (AUTOPILOT-34): set when a runtime failure verdict was * overridden because the item's work is evidenced as LANDED. */ let landedOverrideNote: string | null = null; const store = loadStore(this.cfg.stateDir); if (store) { for (const cand of candidates) { const it = itemByRunId(store, cand); if (it && it.status === "active") { // BUDGET-GOVERNANCE: make the FAIL cause EXPLICIT. A run that was // cut off by its wall-clock budget (the runtime reported timedOut, // or the run lived at/over the item's recorded timeoutMs) is a CAP // failure — the work is unjudged and the recovery is a bigger-budget // re-dispatch — NOT the same as a run that ended with an unsuccessful // verdict. Recorded as the machine flag failCause + a human note. // The store change goes through mutateStore (load → apply → write, // serialized + CAS): a harness flip on a timer used to clobber a // concurrent tool write with its own whole-file snapshot. // FINISHER-EVIDENCE (AUTOPILOT-34): a FINISHER-CLASS run writes into // the declared cwd's checkout, NOT its worktree, so the runtime's // "no edits in the worktree" signal cannot judge it. Before believing // a failure verdict, ask the authoritative record: did the source // THIS run was dispatched to land enter the declared cwd's history // during this run? If it did, the work LANDED — the verdict is // overridden (and the override RECORDED, so a pattern of overrides is // visible), and the item takes the normal success path (ai-review: // the reviewer still verifies the landed commit). Anything short of // that — no declared source, stale/absent baseline, HEAD not advanced // — is no evidence, so the failure stands: the real "wrote a plan and // stopped" case still fails. if (outcome === "failed" && isFinisherItem(it)) { const evidence = finisherLandedEvidence(it, now); if (evidence) { const note = landedNote(evidence, topRunId, now); mutateStore(this.cfg.stateDir, (s) => { const fresh = s.items[it.key]; if (!fresh || fresh.status !== "active") return; updateItem(s, it.key, { status: "ai-review", landedEvidence: evidence, overrides: appendOverride(fresh, landedOverride(evidence, topRunId, now)), notes: fresh.notes ? `${fresh.notes}\n\n${note}` : note, }); }); outcome = "ai-review"; landedOverrideNote = note; flipped = true; flippedKey = it.key; this.logEvent("flip", { key: it.key, runId: cand, outcome, source: "finisher-landed" }); break; } } if (outcome === "failed") { failCause = workerFailCause(ev, it.timeoutMs, it.updatedAt, now); // The two notes below promise harness auto-recovery — which is NOT // offered to a finisher-class item (AUTOPILOT-46: an automatic // re-run of a landing can duplicate a merge), so say so where the // promise is made instead of leaving the operator waiting for it. const heldNote = isFinisherItem(it) ? " FINISHER-CLASS: auto-recovery does NOT re-dispatch this item — it holds it and escalates; re-running the landing is your deliberate act." : ""; // THE PARK (KEY: AUTOPILOT-49) is named where the failure is // recorded, because that is the moment the orchestrator decides // whether a retry can help. Auto-recovery re-dispatches on a 30s // timer and reads STATUS, not prose — so an item that failed on // something no re-run can supply has to be MOVED, not annotated. const parkNote = isFinisherItem(it) ? "" : ` If retrying CANNOT fix this (an interactive prompt no headless run can answer, a missing credential/permission, a decision only the user can make), PARK it instead of leaving it to auto-recovery: queue_update('${it.key}', {status: 'blocked', blocker: 'decision'}) — a blocked item is never auto-re-dispatched. Release it with queue_update('${it.key}', {status: 'approved'}) then queue_dispatch.`; const capNote = failCause === "budget-capped" ? `[budget-capped] ${new Date(now).toISOString()} — worker run ${topRunId} hit the item's wall-clock budget` + (it.timeoutMs ? ` (recorded budget ${formatDurationMs(it.timeoutMs)} = ${it.timeoutMs}ms — the deadline the child ran under, AUTOPILOT-129)` : " (no budget recorded on the item — the runtime itself reported the timeout)") + ` and was CUT OFF mid-task — this is a CAP, not a verdict on the work. Partial work may exist on the pi-parallel-* branch (verify before re-dispatch). RE-DISPATCH WITH A BIGGER BUDGET: queue_update('${it.key}', {timeoutMs: }) then queue_dispatch — the recorded budget rides every dispatch lane (the harness also auto-recovers capped items with a bigger budget).${heldNote}${parkNote}` : failCause === "spawn" ? `[failed: spawn] ${new Date(now).toISOString()} — worker run ${topRunId} died at the provider/infra layer (bare 400 / empty api_error — no deliverable produced), not a verdict on the work. The harness AUTO-RECOVERS with backoff (bounded retries, then escalates). Verify partial work on the pi-parallel-* branch.${heldNote}${parkNote}` : `[failed: verdict] ${new Date(now).toISOString()} — worker run ${topRunId} ended unsuccessfully (not budget-capped). Read its output; verify partial work on the pi-parallel-* branch before re-dispatching (failed items are re-dispatchable: queue_dispatch).${heldNote}${parkNote}`; mutateStore(this.cfg.stateDir, (s) => { const fresh = s.items[it.key]; if (!fresh || fresh.status !== "active") return; updateItem(s, it.key, { status: "failed", failCause, notes: fresh.notes ? `${fresh.notes}\n\n${capNote}` : capNote, }); }); } else { // WORKER SUCCESS → ai-review. Capture the worker's ACCEPTANCE // ARTIFACT here (KEY: AUTOPILOT-108): it is the auditable test // evidence the reviewer will AUDIT instead of re-running the suite. // Absent → no field is written; the review task then states plainly // that no artifact was captured and the reviewer must record the // load-bearing claims as COULD NOT CHECK. const evidence = parseAcceptanceEvidence(ev, this.cfg.workerAgents); mutateStore(this.cfg.stateDir, (s) => { if (s.items[it.key]?.status === "active") { updateItem(s, it.key, { status: "ai-review", ...(evidence ? { acceptanceEvidence: storedAcceptanceEvidence(evidence, new Date(now).toISOString()) } : {}), }); } }); } flipped = true; flippedKey = it.key; // capture inside the loop — the FIRST matching item, not any same-status item this.logEvent("flip", { key: it.key, runId: cand, outcome, failCause: failCause ?? undefined }); break; } } } // A flipped item (or a ledger-tracked run) freed a worker slot. const freedSlot = flipped || entry !== undefined; const isReviewerRun = [ev.agent, ...(Array.isArray(ev.results) ? ev.results.map((r) => (r as { agent?: string })?.agent) : [])] .filter((a): a is string => typeof a === "string" && !!a) .some((a) => this.cfg.reviewerAgents.includes(a)); const domainEvents: DomainEvent[] = []; if (flipped) { domainEvents.push({ name: "orch:item-completed", data: { key: flippedKey, runId: topRunId, outcome } }); // A WORKER FAILURE must never be silent (the historical gap: children // died at the budget default with no signal; recovery was archaeology). // Surface it as a failure tick — the operator sees it was capped and the // message routes a bigger-budget re-dispatch instead of generic // fail-forward. The verdict path also ticks (a failed run is exactly // when judgment is needed), without the budget-cap wording. if (outcome === "failed") { const item = loadStore(this.cfg.stateDir)?.items[flippedKey]; const cap = item?.timeoutMs ?? null; domainEvents.push({ name: "orch:item-failed", data: { key: flippedKey, runId: topRunId, cause: failCause, timeoutMs: cap } }); const capped = failCause === "budget-capped"; const infra = failCause === "spawn"; return { tick: { reason: "failure", message: capped ? `[orch-tick: failure] ${flippedKey} FAILED — its worker run ${topRunId.slice(0, 8)} hit the item's wall-clock budget cap` + (cap ? ` (${formatDurationMs(cap)})` : "") + ` and was CUT OFF mid-task (budget-capped, NOT a verdict on the work). Verify partial work on the pi-parallel-* branch, then RE-DISPATCH WITH A BIGGER BUDGET: queue_update('${flippedKey}', {timeoutMs: ${cap ? `> ${cap}` : ""}}) then queue_dispatch (or let auto-dispatch take it — the recorded budget rides every lane). Respond ≤2 lines.` : infra ? `[orch-tick: failure] ${flippedKey} FAILED — its worker run ${topRunId.slice(0, 8)} died at the provider/infra layer (bare 400 / empty api_error, no deliverable) — NOT a verdict on the work. The harness AUTO-RECOVERS with a short backoff (bounded retries, then escalates); verify partial work on the pi-parallel-* branch. Respond ≤2 lines.` : `[orch-tick: failure] ${flippedKey} FAILED — its worker run ${topRunId.slice(0, 8)} ended unsuccessfully (verdict/exit, not budget-capped). Read the run's output and verify partial work on the pi-parallel-* branch before re-dispatching (failed items are re-dispatchable: queue_dispatch). Respond ≤2 lines.`, facts: { key: flippedKey, outcome: "failed", failCause, budgetCapped: capped, ...(cap !== null ? { timeoutMs: cap } : {}), }, }, domainEvents, flipped, freedSlot, reviewerCompleted: isReviewerRun, }; } // A FALSE FAILURE THAT THE QUEUE OVERRODE must be visible: the operator // just saw the runtime report the run as failed, while the queue moved // the item forward on landed evidence. Silence there is how the two // views drift apart (and how a real failure eventually gets waved // through). The override itself is already RECORDED on the item. if (landedOverrideNote) { const evidence = loadStore(this.cfg.stateDir)?.items[flippedKey]?.landedEvidence ?? null; domainEvents.push({ name: "orch:failure-overridden", data: { key: flippedKey, runId: topRunId, evidence } }); return { tick: { reason: "review", message: `[orch-tick: review] ${flippedKey} — the runtime reported run ${topRunId.slice(0, 8)} UNSUCCESSFUL, but this is a FINISHER-CLASS dispatch (writes outside its worktree) and the source it was sent to land IS IN: ` + `${evidence ? `${evidence.source ?? evidence.sourceSha.slice(0, 8)} → ${evidence.repo} ${evidence.ref ?? "HEAD"} ${String(evidence.fromSha ?? "?").slice(0, 8)} → ${evidence.sha.slice(0, 8)}` : "the declared source landed in the declared cwd"}. ` + `The failure verdict was OVERRIDDEN (recorded in the item's overrides[]) and ${flippedKey} moved to ai-review — do NOT re-dispatch it; verify the landed commit. Respond ≤2 lines.`, facts: { key: flippedKey, outcome: "ai-review", failureOverridden: true, ...(evidence ? { landedSha: evidence.sha, repo: evidence.repo } : {}) }, }, domainEvents, flipped, freedSlot, reviewerCompleted: isReviewerRun, }; } return { tick: null, domainEvents, flipped, freedSlot, reviewerCompleted: isReviewerRun }; } // REVIEWER completion path: attribute to the item via reviewerRunId, parse // the verdict line, auto-flip (strict parse only), and tick the orchestrator // for the remaining judgment (flag_for_review / re-dispatch / cap surface). if (isReviewerRun && store) { // ATTRIBUTION (KEY: AUTOPILOT-104): the primary key is the reviewer run // recorded on the item at DISPATCH time (`reviewerRunId`). That record can // legitimately be missing — a reviewer dispatched before its item reached // `ai-review` has the ref cleared by the entering-ai-review stamp, and // requiring it alone silently dropped a genuine PASS (the live ACTUATOR-78 // wedge, repaired by hand). Fall back to the item the run's OWN task names // (`ev.itemKey`, the `KEY: ` line every harness task starts with), // only when the ai-review item carries NO reviewer ref: a DIFFERENT ref is // ambiguous (another round's reviewer) and must surface, never be guessed // through. const matched = candidates.map((c) => itemByReviewerRunId(store, c)).find(Boolean) ?? itemByTaskKey(store, ev.itemKey) ?? null; if (matched) { domainEvents.push({ name: "orch:reviewer-completed", data: { key: matched.key, reviewerRunId: matched.reviewerRunId ?? topRunId } }); const verdict = parseVerdict(ev, this.cfg.reviewerAgents); // CAVEAT CAPTURE (AUTOPILOT-27): the verdict LINE routes the item; the // rest of the reviewer's body used to be discarded here, so the human // approved with less information than the review produced. Capture is // ADDITIVE and total (never throws, degrades to null) — it runs after // the verdict is parsed and cannot influence the flip. const reviewNotes = verdict ? extractReviewNotes(ev, this.cfg.reviewerAgents, verdict, new Date(now).toISOString()) : null; // REVIEW-CONTRACT FIELDS (KEY: AUTOPILOT-108/114): the reviewer's // report MUST carry the coverage statement (AUTOPILOT-108) AND, since // AUTOPILOT-114, the verdict reason, a FAIL/NOTE disposition for every // finding (with a per-note justification) and — on a FAIL — the "do not // rebuild" statement. Their absence is a VALIDATION FAILURE — recorded // on the event and surfaced in the tick, never silently accepted. It // does NOT by itself flip the verdict (PASS/FAIL/NO-CHANGE semantics are // unchanged); it makes the report's limit VISIBLE. const contractDeficiencies = verdict ? validateReviewReport(verdictBody(ev, this.cfg.reviewerAgents) ?? "") : []; const contractGap = contractDeficiencies.length > 0; if (contractGap) { domainEvents.push({ name: "orch:review-contract-missing", data: { key: matched.key, verdict, deficiencies: contractDeficiencies } }); this.logEvent("review-contract-missing", { key: matched.key, count: contractDeficiencies.length }); } const contractNote = contractGap ? `\n⚠ REVIEW CONTRACT: this report is missing required fields (${contractDeficiencies[0]}) — it is not an accepted review until the verdict reason and the per-finding territory classification are stated.` : ""; const withinQuiet = now - this.lastReviewTickAt < this.cfg.quietPeriodMs; if (verdict === "PASS") { // AI review passed → the item enters HUMAN review (your approval), // not done. The harness auto-flags it; YOU make it done. mutateStore(this.cfg.stateDir, (s) => { if (s.items[matched.key]) updateItem(s, matched.key, { status: "human-review", reviewNotes }); }); domainEvents.push({ name: "orch:verdict", data: { key: matched.key, verdict: "PASS", attempts: matched.attempts ?? 0 } }); this.logEvent("flip", { key: matched.key, outcome: "human-review", source: "verdict-pass" }); if (!withinQuiet) this.lastReviewTickAt = now; return { tick: withinQuiet ? null : { reason: "review", message: `[orch-tick: review] ${matched.key} PASSED AI review and is in HUMAN review, awaiting you — the harness auto-flagged it (with targets). Approve it (queue_update status: done), re-dispatch (active), or drop (rejected). Not a user request; respond ≤2 lines.${contractNote}`, facts: { key: matched.key, verdict: "PASS", ...(contractGap ? { contractMissing: true } : {}) }, }, domainEvents, flipped: true, freedSlot: true, // reviewer slot freed — dispatch sweep runs too reviewerCompleted: true, }; } if (verdict === "FAIL") { const attempts = (matched.attempts ?? 0) + 1; if (attempts >= this.cfg.reviewCap) { const causeNote = `[failed: verdict] ${new Date(now).toISOString()} — review FAIL at attempt ${attempts} (cap ${this.cfg.reviewCap}) — the work failed AI review ${attempts} times, not a budget cap. Apply the findings directly, re-scope, or drop.`; mutateStore(this.cfg.stateDir, (s) => { const fresh = s.items[matched.key]; if (!fresh) return; updateItem(s, matched.key, { status: "failed", attempts, failCause: "verdict", reviewNotes, notes: fresh.notes ? `${fresh.notes}\n\n${causeNote}` : causeNote, }); }); domainEvents.push({ name: "orch:verdict", data: { key: matched.key, verdict: "FAIL", attempts } }); this.logEvent("flip", { key: matched.key, outcome: "failed", source: "verdict-cap" }); if (!withinQuiet) this.lastReviewTickAt = now; return { tick: withinQuiet ? null : { reason: "review", message: `[orch-tick: review] ${matched.key} FAILED review at attempt ${attempts} (cap ${this.cfg.reviewCap}) — surface to the user: (a) apply findings directly, (b) review as-is, (c) drop. Not a user request; respond ≤2 lines.${contractNote}`, facts: { key: matched.key, verdict: "FAIL", attempts, cap: this.cfg.reviewCap, ...(contractGap ? { contractMissing: true } : {}) }, }, domainEvents, flipped: true, freedSlot: true, // reviewer slot freed — dispatch sweep runs too reviewerCompleted: true, }; } mutateStore(this.cfg.stateDir, (s) => { if (s.items[matched.key]) updateItem(s, matched.key, { status: "active", attempts, runId: null, reviewerRunId: null, reviewNotes }); }); domainEvents.push({ name: "orch:verdict", data: { key: matched.key, verdict: "FAIL", attempts } }); this.logEvent("flip", { key: matched.key, outcome: "active", source: "verdict-fail", attempts }); if (!withinQuiet) this.lastReviewTickAt = now; return { tick: withinQuiet ? null : { reason: "review", message: `[orch-tick: review] ${matched.key} FAILED review (attempt ${attempts}) — the item is ACTIVE: the harness auto-re-dispatches it (queue_dispatch only if autopilot's auto-dispatch is off). Before the redo runs: VERIFY GROUND TRUTH — false-negative reviews are the recurring pattern (the reviewer may have missed the work on a parallel branch); if the deliverable is already delivered, stop the redo + reconcile the item. Not a user request; respond ≤2 lines.${contractNote}`, facts: { key: matched.key, verdict: "FAIL", attempts, ...(contractGap ? { contractMissing: true } : {}) }, }, domainEvents, flipped: true, freedSlot: true, // reviewer slot freed — dispatch sweep runs too reviewerCompleted: true, }; } if (verdict === "NO-CHANGE") { // NON-DELIVERABLE OUTCOME (KEY: AUTOPILOT-108): the reviewer ruled that // the CORRECT outcome is NO change — a refusal/withdrawal with no // artifact. PASS/FAIL cannot express it: PASS implies an artifact was // produced and accepted, and FAIL RE-DISPATCHES (which is how a correct // refusal would have been re-run into a destructive change; only a // supervisor reply routed it safely). TERMINAL-WITH-RECORD: the item // moves to `no-change` — never auto-re-dispatched (auto-dispatch only // touches `approved`, auto-recovery only `failed`) — and the reviewer's // rationale is captured in reviewNotes. No artifact is accepted or // produced. mutateStore(this.cfg.stateDir, (s) => { if (s.items[matched.key]) updateItem(s, matched.key, { status: "no-change", reviewNotes }); }); domainEvents.push({ name: "orch:verdict", data: { key: matched.key, verdict: "NO-CHANGE" } }); this.logEvent("flip", { key: matched.key, outcome: "no-change", source: "verdict-no-change" }); if (!withinQuiet) this.lastReviewTickAt = now; return { tick: withinQuiet ? null : { reason: "review", message: `[orch-tick: review] ${matched.key} — the reviewer's verdict is NO-CHANGE: the correct outcome was to make NO change (a refusal/withdrawal; no deliverable). The item is CLOSED as 'no-change' (terminal, recorded) and will NOT be re-dispatched. Read the reviewer's rationale; if the premise is wrong, re-open deliberately (queue_update status: approved) and dispatch. Not a user request; respond ≤2 lines.${contractNote}`, facts: { key: matched.key, verdict: "NO-CHANGE", terminal: true, redispatch: false, ...(contractGap ? { contractMissing: true } : {}) }, }, domainEvents, flipped: true, freedSlot: true, // reviewer slot freed — dispatch sweep runs too reviewerCompleted: true, }; } // no parseable verdict → manual path. This is a ROUTING FAILURE, not // a resting state (KEY: AUTOPILOT-99): the item would otherwise sit // in ai-review with no machine signal, so record an explicit event // naming the item carrying the tail of the reviewer's own words for // adjudication — the operator must not have to open artifacts by hand. const aiReviewing = Object.values(store.items).filter((i) => i.status === "ai-review").map((i) => i.key); const reviewerRunId = matched.reviewerRunId ?? topRunId; const tail = verdictTail(ev, this.cfg.reviewerAgents); domainEvents.push({ name: "orch:verdict-unparseable", data: { key: matched.key, reviewerRunId, tail } }); this.logEvent("verdict-unparsed", { key: matched.key, reviewerRunId, tailLength: tail?.length ?? 0 }); return { tick: { reason: "review", message: `[orch-tick: review] Reviewer for ${matched.key} completed but the verdict line was not parseable — the item is STALLED in ai-review (no status change). Adjudicate from the reviewer's tail below (or run ${reviewerRunId.slice(0, 8)}'s output): move it to human-review / re-dispatch / failed yourself.` + `${tail ? `\n--- reviewer output (tail) ---\n${tail}` : ""}` + `\nIn ai-review: ${aiReviewing.join(", ") || "none"}. Not a user request; respond ≤2 lines.`, facts: { key: matched.key, aiReview: aiReviewing, verdictUnparseable: true }, }, domainEvents, flipped: false, freedSlot: true, // reviewer slot freed — dispatch sweep runs too reviewerCompleted: true, }; } // No item could be attributed to this reviewer completion. A VERDICT the // reviewer actually rendered must NEVER be a silent no-op (KEY: // AUTOPILOT-104): record an explicit event + attention tick naming the // run, the verdict and the candidates considered, so a real PASS/FAIL // cannot be lost without a trace. A completion with NO verdict (an ad-hoc // plain subagent review not tied to a queue item) keeps the generic tick. const orphanVerdict = parseVerdict(ev, this.cfg.reviewerAgents); if (orphanVerdict) { const tail = verdictTail(ev, this.cfg.reviewerAgents); const aiReviewing = Object.values(store.items).filter((i) => i.status === "ai-review").map((i) => i.key); domainEvents.push({ name: "orch:verdict-unattributed", data: { runId: topRunId, agent: ev.agent ?? null, verdict: orphanVerdict, candidates, itemKey: ev.itemKey ?? null, tail }, }); this.logEvent("verdict-unattributed", { runId: topRunId, verdict: orphanVerdict, candidates, itemKey: ev.itemKey ?? null }); return { tick: { reason: "review", message: `[orch-tick: review] REVIEWER VERDICT UNATTRIBUTED — reviewer run ${topRunId.slice(0, 8)} completed with ${orphanVerdict} but no ai-review item carries its run id` + `${ev.itemKey ? ` (its task names ${ev.itemKey}, but that item is not awaiting review with no reviewer ref)` : ""} — this is a ROUTING FAILURE, never a silent no-op. ` + `Find the item this run reviewed and route it yourself (PASS → human-review, FAIL → re-dispatch), or surface it to the user.` + `${tail ? `\n--- reviewer output (tail) ---\n${tail}` : ""}` + `\nIn ai-review: ${aiReviewing.join(", ") || "none"}. Not a user request; respond ≤2 lines.`, facts: { runId: topRunId, verdict: orphanVerdict, candidates, verdictUnattributed: true }, }, domainEvents, flipped: false, freedSlot: true, // reviewer slot freed — dispatch sweep runs too reviewerCompleted: true, }; } // reviewer completed with no verdict and no attributed item (a plain // ad-hoc subagent review) → generic review tick const generic = this.reviewTick(now); return { tick: generic, domainEvents, flipped: false, freedSlot: true, // reviewer slot freed — dispatch sweep runs too reviewerCompleted: true, }; } // SPEC-PASS completion path (KEY: AUTOPILOT-36): a spec-pass worker // (dispatched by the panel's `spec-pass` action — see // dispatchPendingSpecPass, framework/auto-dispatch.ts) is READ-ONLY // reconnaissance against ONE proposal. It never touches `active`/ // `ai-review`, so neither branch above matches it — attribution is by its // OWN field, `specPassRunId`, against a `proposal` item ONLY. // NEVER APPROVES: this writes `scope` back and clears `specPassRunId` — // status is untouched either way (still `proposal`), so approval stays // exactly as human-only as it was before the pass ran. An item that // moved on (approved/rejected/deferred) WHILE the pass was running // already had its `specPassRunId` cleared by updateItem's leaving- // proposal stamping (queue-store.ts) — `itemBySpecPassRunId` is scoped to // `proposal`, so a late completion for a superseded item finds nothing to // attribute to and falls through to the unmatched-completion telemetry // below, the SAME "item moved on, no orphaned write" guarantee // AUTOPILOT-59's claim-before-spawn gives human re-dispatch. if (!flipped && store) { const specMatch = candidates.map((c) => itemBySpecPassRunId(store, c)).find(Boolean) ?? null; if (specMatch) { domainEvents.push({ name: "orch:spec-pass-completed", data: { key: specMatch.key, runId: topRunId } }); const completedScope = parseSpecPassScope(ev, this.cfg.workerAgents); const nowIso = new Date(now).toISOString(); if (completedScope) { mutateStore(this.cfg.stateDir, (s) => { const fresh = s.items[specMatch.key]; if (!fresh || fresh.status !== "proposal") return; // moved on — nothing to write back to const note = `[spec pass] ${nowIso} — scope completed by run ${topRunId}; still a proposal, approval is still yours.`; updateItem(s, specMatch.key, { scope: completedScope, specPassRunId: null, notes: fresh.notes ? `${fresh.notes}\n\n${note}` : note, }); }); this.logEvent("flip", { key: specMatch.key, outcome: "proposal", source: "spec-pass" }); return { tick: { reason: "review", message: `[orch-tick: review] spec pass for ${specMatch.key} completed — its scope was filled in (still a PROPOSAL; nothing here approved it). Read it and decide: approve, refine further, or reject. Not a user request; respond ≤ 2 lines.`, facts: { key: specMatch.key, specPass: "completed" }, }, domainEvents, flipped: false, freedSlot: true, reviewerCompleted: false, }; } // Output didn't follow the contract — clear the run ref, leave the // scope untouched, and say so plainly rather than guess. mutateStore(this.cfg.stateDir, (s) => { const fresh = s.items[specMatch.key]; if (!fresh || fresh.status !== "proposal") return; const note = `[spec pass] ${nowIso} — run ${topRunId} completed but its output did not carry a parseable completed scope (missing the marker block) — the scope was NOT changed; read its output directly or retry.`; updateItem(s, specMatch.key, { specPassRunId: null, notes: fresh.notes ? `${fresh.notes}\n\n${note}` : note }); }); this.logEvent("flip", { key: specMatch.key, outcome: "proposal", source: "spec-pass-unparsed" }); return { tick: { reason: "review", message: `[orch-tick: review] spec pass for ${specMatch.key} completed but its output was not parseable — the scope was NOT changed. Read run ${topRunId.slice(0, 8)}'s output yourself. Not a user request; respond ≤ 2 lines.`, facts: { key: specMatch.key, specPass: "unparsed" }, }, domainEvents, flipped: false, freedSlot: true, reviewerCompleted: false, }; } } // UNMATCHED COMPLETION telemetry: a worker/reviewer run ended but no // store item carries its id. This is the signature of a pipeline bypass // (an ad-hoc `subagent` spawn that never went through queue_dispatch, so // no run linkage was ever written) — the work lands on some branch while // the item stays put, and WITHOUT this line the only diagnosis is // archaeology across session logs (observed live: EVAL-WATCHDOG-ACTIVITY). // Non-harness agents (scouts, ad-hoc Q&A) are expected to be unmatched — // don't log those. if (!flipped && !isReviewerRun && (this.cfg.workerAgents.includes(ev.agent ?? "") || this.cfg.reviewerAgents.includes(ev.agent ?? ""))) { this.logEvent("unmatched-completion", { runId: topRunId, agent: ev.agent, candidates }); } return { tick: null, domainEvents, flipped, freedSlot, reviewerCompleted: isReviewerRun }; } /** agent_settled → the orchestrator's own turn ended; check for unacted capacity. */ handleAgentSettled(now = this.now()): AutopilotResult { return this.sweep("settled", now); } /** * Explicit capacity sweep — activation (/autopilot on), periodic timer, * settled turns, and store-driven completions. `occupied` is the UNION of * the authoritative fleet count (pi-subagents fleet status) and the store's * own live-item derivation (liveTrackedSlots) — one derivation shared with * the queue tools, so the tick's FLEET line can never contradict them * (KEY: AUTOPILOT-86). */ sweep(source: "settled" | "activate" | "timer" | "worker-done", now = this.now(), fleet?: { totalActive?: number; occupied?: number }, backend?: Pick | null): AutopilotResult { const snapshot = this.readQueueSnapshot(now, backend); // Occupied = ALL subagents dispatched from the orchestrator session — // workers, reviewers, scouts, plain subagent calls (the general case). // - fleet.totalActive is the AUTHORITATIVE count (the backend's // FLEET-vs-INVENTORY parity union: the pi-subagents status RPC + the // in-flight async-run inventory — the same source `subagent status` // reads). It sees every run this backend spawns, whatever its role, // including parents the session-gated RPC alone would undercount. // - FLEET-vs-INVENTORY PARITY (AUTOPILOT-6): the fleet count is the // FLOOR; the delivery-time numbers are the store's own LIVE-item // derivation (snapshot.occupied — liveTrackedSlots, AUTOPILOT-86). // // KEY: AUTOPILOT-71 — this used to ALSO take `this.ledger.size` (the // engine's run ledger, a Map maintained by handleAsyncStarted / // handleAsyncComplete) into the max, the SAME leaked-forever ledger // AUTOPILOT-61/68 already banned from effectiveOccupiedSlots // (auto-dispatch.ts) and runner.ts's effectiveTotalActive — a ledger // entry whose completion event never arrives with a matching id leaks // forever and permanently inflates this reading regardless of what the // fleet or store actually say is running. Same fix shape, same reason: // drop the ledger term, trust only snapshot.occupied (store) unioned // with the authoritative fleet count. // // ACCEPTED TRADEOFF: three tests (autopilot.test.ts) previously exercised // this callsite's ledger fallback as a DELIBERATE undercount defense — // when the store forgets to record a dispatch (or an ad-hoc/scout run // the queue never tracks at all) the ledger caught it. That defense is // now gone HERE too: a scout/ad-hoc run that never touches the store is // invisible to sweep()/refreshTickFacts() occupied/tick text, exactly as // it already is invisible to effectiveOccupiedSlots/effectiveTotalActive. // The fleet RPC (fleet?.totalActive) remains the other half of the // union and is the intended backstop for that case when available. // // ONE OCCUPANCY (KEY: AUTOPILOT-96): `fleet.occupied` is the GATE's // occupancy for this pass — the runner computes effectiveOccupiedSlots // (the max of the store's live items and the authoritative run-level // fleet count) ONCE and passes it here. The tick then PROJECTS that // number: it can never be lower than what the gate acted on, and it is // unioned with the store's fresh live-item count so slots the gate just // filled are still counted. Before this, the delivery-time refresh was // handed NO fleet term at all, recomputing from the store alone and // reporting free slots the gate did not have — how a held tick could read // "3/6, 3 free" while the gate sat at cap. The bare union below is the // engine-only fallback (no gate in the caller's pass). const tracked = snapshot.occupied; const occupied = Math.max(fleet?.occupied ?? 0, fleet?.totalActive ?? 0, tracked); const eff = { ...snapshot, occupied, tracked }; const hash = this.queueHash(eff); // settled/activate/worker-done: only tick when the queue state changed. // timer: re-nudge a PERSISTENT gap even when unchanged — the orchestrator // may have ignored the earlier nudge (or was mid-flight), and the 10-min // interval is the throttle. Without this, a gap nudged once and unacted // is never nudged again. if (source !== "timer" && hash === this.lastSettledHash) return empty(); this.lastSettledHash = hash; this.logEvent("sweep", { source, occupied, ready: eff.ready, ledger: this.ledger.size, fleetTotalActive: fleet?.totalActive }); const tick = this.decideTick(eff, source, undefined, now, true, fleet?.totalActive); if (tick) { this.logEvent("tick", { reason: tick.reason, source }); return { tick, domainEvents: [], flipped: false, freedSlot: false, reviewerCompleted: false }; } // BUDGET TELEMETRY at the timer heartbeat: an ACTIVE budgeted run past // ~75% of its wall-clock budget gets ONE proactive warning per run (the // operator can steer it to wrap up / commit before the cap cuts it off) // instead of the failure being the first signal. Nudged on the timer only // (the completion path for the cap itself is event-driven + deduped); the // warned set prevents a 10-minute nag for the same run. if (source === "timer") { const budgetTick = this.budgetWarningTick(eff.budget); if (budgetTick) { this.logEvent("tick", { reason: budgetTick.reason, source }); return { tick: budgetTick, domainEvents: [], flipped: false, freedSlot: false, reviewerCompleted: false }; } } return empty(); } /** * Zombie reconciliation — the deterministic safety net for LOST completion * events. The active→ai-review/failed flip is event-driven * (subagent:async-complete on the spawning session's in-process bus), so a * run that ends while no session is listening (timeout overnight, crash, * restart) leaves its item active FOREVER — observed live (EVAL-EXPT-M3: * 8h timeout, still active the next day). The AUTHORITATIVE counter-evidence * is fleetStatus: when it reports zero active runs, NOTHING is running * anywhere this backend spawned (the backend count now unions the RPC with * the in-flight async inventory — an undercounting RPC can no longer fake a * 0, AUTOPILOT-6) — any `active` item idle past the grace * window is a zombie by definition. Flip it to failed with the evidence in * notes; the orchestrator then re-dispatches deliberately (verifying * pi-parallel-* branches for partial work first) instead of a dead run * occupying a phantom slot indefinitely. * * FAKE-ZERO GUARD (AUTOPILOT-6): only an AUTHORITATIVE fleet count may * trigger the net FOR AN ITEM WITH A REAL RUN ID. When the RPC failed / the * backend was unreachable the runner passes `undefined` — that is NOT * evidence of an idle fleet and must never be coerced into a 0 (the old * `?? 0` could flip LIVE runs whose status call just timed out). Undefined * or non-zero → a runId-bearing item is left alone; a real 0 is required. * * NULL-RUNID ITEMS ARE A DIFFERENT SHAPE (KEY: AUTOPILOT-64) — the fleet-idle * gate above exists to protect a run that MIGHT be one of the fleet's busy * slots; an `active` item with NO recorded runId has no run anywhere to be * one of those slots (the review-FAIL retry path nulls runId and relies on * autoRedispatch to repopulate it — see auto-dispatch.ts; a spawn failure * there can leave it null). Gating that shape behind total-fleet-idleness * conflates "no live run for THIS item" with "no live run for ANYTHING" — * it only got caught in practice once the entire fleet happened to go idle * (rare mid-session), observed live on AUTOPILOT-42 sitting stuck for hours * while other workers kept the fleet busy. A null-runId item costs nothing * extra to check — there is no run to look up — so it is checked on EVERY * pass regardless of `fleetTotalActive`, past the same grace window. */ zombieReconcile(fleetTotalActive: number | undefined, now = this.now()): { flippedKeys: string[] } { const graceMs = this.cfg.zombieGraceMinutes * 60_000; if (graceMs <= 0) return { flippedKeys: [] }; // 0 disables the sweep if (!loadStore(this.cfg.stateDir)) return { flippedKeys: [] }; const fleetAuthoritativelyIdle = fleetTotalActive === 0; // The whole scan runs INSIDE the mutation: deciding from one snapshot and // writing a different one is the lost-update window. mutateStore may re-run // this on a conflict, so it stays a pure function of the store it is given // (the logging happens after it returns). const flipped = mutateStore(this.cfg.stateDir, (store) => { const keys: Array<{ key: string; runId: string | null }> = []; for (const it of Object.values(store.items)) { if (it.status !== "active") continue; // A real runId still needs the fleet AUTHORITATIVELY idle (fake-zero // guard); a null runId has nothing for a busy fleet to be running, so // it is checked unconditionally. if (it.runId && !fleetAuthoritativelyIdle) continue; const updatedAt = Date.parse(it.updatedAt); const idleMs = now - updatedAt; if (!Number.isFinite(updatedAt) || idleMs < graceMs) continue; const evidence = it.runId ? `zombie reconciliation ${new Date(now).toISOString()} — fleet reported 0 active runs while this item sat active ~${Math.round(idleMs / 60_000)}m past update ` + `(run ${it.runId}): timeout/crash/lost completion event. Partial work may exist on pi-parallel-* branches — verify before re-dispatch.` : `zombie reconciliation ${new Date(now).toISOString()} — this item has been active with NO recorded run id for ~${Math.round(idleMs / 60_000)}m ` + `(a re-dispatch spawn never repopulated it — see the item's notes for the spawn failure, if any). Checked every sweep regardless of fleet ` + `occupancy: there is no run anywhere for a busy fleet to be running on this item's behalf. Partial work, if any, may exist on pi-parallel-* branches — verify before re-dispatch.`; updateItem(store, it.key, { status: "failed", failCause: "zombie", notes: it.notes ? `${it.notes}\n\n${evidence}` : evidence, }); keys.push({ key: it.key, runId: it.runId ?? null }); } return keys; }); // LEDGER RELEASE (AUTOPILOT-54): the whole point of this net is a LOST // completion event — handleAsyncComplete (the only other ledger writer) // will never run for this run, so its ledger entry is deleted here or it // never is. Without this, fleet.occupied (ledger.size) only ever grows: // observed live climbing 12→13→15→17 while the true live-process count // (this very fleetTotalActive===0 precondition) was zero every time. for (const f of flipped) if (f.runId) this.ledger.delete(f.runId); for (const f of flipped) this.logEvent("flip", { key: f.key, runId: f.runId, outcome: "failed", source: "zombie" }); return { flippedKeys: flipped.map((f) => f.key) }; } /** * Dead-run reconciliation — the PER-RUN half of the safety net (AUTOPILOT-47). * * WHY zombieReconcile ALONE COULD NEVER CATCH THIS: it is gated on the fleet * being AUTHORITATIVELY IDLE (`fleetTotalActive === 0`). That gate exists for * a good reason — without a per-run signal, "nothing is running anywhere" was * the only evidence strong enough to condemn an item. But it makes the net * IMPOSSIBLE TO FIRE on a busy queue: one live worker anywhere keeps the count * above zero and every dead run in the store is immune. Observed live: three * runs sat dead-but-`active`/`ai-review` for 117, 287 and more minutes while * other workers ran, and every item gated behind them waited on a corpse. * * The per-run evidence is stronger than the fleet count anyway: the run's OWN * status.json says `state:"failed"`. This sweep uses exactly that and nothing * weaker — see `runStateEvidence` for why `terminal` and `undeterminable` must * stay distinct, and why the fail-open direction here is the OPPOSITE of the * reviewer lane's. `undeterminable` NEVER flips: no backend, a missing run * dir, an unreadable status.json, or an unrecognised phase all leave the item * exactly where it is. Only a status file that parsed and named a terminal * phase condemns a run. * * The grace window is the SAME `zombieGraceMinutes` the fleet-idle net uses, * for the same reason: the active→ai-review flip is event-driven, so an item * whose run just ended is legitimately `active` for a moment. Only an item * still `active` a full grace window after its last update has provably lost * its completion event. */ deadRunReconcile( runEvidence: (runId: string) => { kind: "in-flight" | "terminal" | "undeterminable"; phase?: string; error?: string }, now = this.now(), ): { flippedKeys: string[] } { const graceMs = this.cfg.zombieGraceMinutes * 60_000; if (graceMs <= 0) return { flippedKeys: [] }; // 0 disables the sweep, as for zombieReconcile if (!loadStore(this.cfg.stateDir)) return { flippedKeys: [] }; // Evidence is gathered OUTSIDE the mutation (it touches the filesystem and // mutateStore may re-run its apply on a conflict); the apply below stays a // pure function of the store plus this snapshot. const store = loadStore(this.cfg.stateDir); if (!store) return { flippedKeys: [] }; const condemned = new Map(); for (const it of Object.values(store.items)) { if (it.status !== "active" || !it.runId) continue; const updatedAt = Date.parse(it.updatedAt); const idleMs = now - updatedAt; if (!Number.isFinite(updatedAt) || idleMs < graceMs) continue; let ev: { kind: string; phase?: string; error?: string }; try { ev = runEvidence(it.runId); } catch { continue; // the evidence lookup itself failed — undeterminable, never death } if (ev.kind !== "terminal") continue; // in-flight OR undeterminable → leave it alone condemned.set(it.key, { runId: it.runId, phase: ev.phase ?? "terminal", error: ev.error ?? null, idleMs }); } if (condemned.size === 0) return { flippedKeys: [] }; const flipped = mutateStore(this.cfg.stateDir, (s) => { const keys: Array<{ key: string; runId: string }> = []; for (const [key, c] of condemned) { const it = s.items[key]; // Re-check under the mutation: the completion event may have landed // between the snapshot and the write, and the run id must still match. if (!it || it.status !== "active" || it.runId !== c.runId) continue; const evidence = `dead-run reconciliation ${new Date(now).toISOString()} — run ${c.runId} reports state '${c.phase}' in its own status.json ` + `while this item sat active ~${Math.round(c.idleMs / 60_000)}m past update: the completion event was lost.` + (c.error ? ` Run error: ${c.error}` : "") + ` Partial work may exist on pi-parallel-* branches — verify before re-dispatch.` + // AUTOPILOT-47 C: UNCOMMITTED work survives only in the run's worktree // directory; naming it here is the difference between a mechanical // salvage and matching run ids to pi-worktree-* paths by hand. (it.runWorktree?.path ? ` Uncommitted work (if any) is in the run's worktree: ${it.runWorktree.path}` : ""); updateItem(s, key, { status: "failed", failCause: deadRunFailCause(c.error), notes: it.notes ? `${it.notes}\n\n${evidence}` : evidence, }); keys.push({ key, runId: c.runId }); } return keys; }); // LEDGER RELEASE (AUTOPILOT-54): same reasoning as zombieReconcile — this // net exists BECAUSE the completion event was lost, so handleAsyncComplete // will never delete this run's ledger entry. Release it here or fleet.occupied // leaks it forever. for (const f of flipped) this.ledger.delete(f.runId); for (const f of flipped) this.logEvent("flip", { key: f.key, runId: f.runId, outcome: "failed", source: "dead-run" }); return { flippedKeys: flipped.map((f) => f.key) }; } /** * Review tick — the deterministic trigger for the orchestrator's judgment * step: when items are stuck in `reviewing`, the orchestrator has no other * way to know a reviewer finished. Fired on reviewer completion and by the * periodic timer. The verdict (PASS/FAIL/NO-CHANGE) is the orchestrator's call — the * tick only says "route it". * * STUCK-TICK LIVENESS FILTER (extends AUTOPILOT-47's terminal-status * distinction to this safety net; AUTOPILOT-50). The periodic `"stuck"` tick * used to fire the FULL "check each" procedure text whenever ANY item sat in * `ai-review`, even when every one of them had a reviewer PROVABLY still * running — observed 5+ times in one session, including a fire while * autopilot was OFF. An item whose reviewer is confirmed `in-flight` needs no * check: "its completion will notify" is already true, so naming it just * spends an orchestrator turn on nothing. `runEvidence` (the same seam * `deadRunReconcile` uses) lets the caller supply that confirmation; items * WITHOUT it — no reviewer dispatched yet, a `terminal` run (the reviewer's * own completion event was lost, same shape as the dead-run sweep), or * `undeterminable` (AUTOPILOT-20's fail-OPEN boundary: never read as proof of * death) — stay in the nudge. No `runEvidence` supplied (the * `"reviewer-completed"` caller, and any `"stuck"` caller that cannot resolve * run dirs) keeps the old unfiltered behaviour — never a NEW false silence. * When filtering leaves nothing to check and no human-review is pending, the * tick is silent (`null`) instead of firing an empty procedure. */ reviewTick( now = this.now(), why: "reviewer-completed" | "stuck" = "reviewer-completed", runEvidence?: (runId: string) => { kind: "in-flight" | "terminal" | "undeterminable" }, ): Tick | null { const store = loadStore(this.cfg.stateDir); if (!store) return null; const aiReview = Object.values(store.items).filter((i) => i.status === "ai-review"); const humanReview = Object.values(store.items).filter((i) => i.status === "human-review"); if (!aiReview.length && !humanReview.length) return null; const filterStuck = why === "stuck" && !!runEvidence; const checkable = filterStuck ? aiReview.filter((i) => !i.reviewerRunId || runEvidence!(i.reviewerRunId).kind !== "in-flight") : aiReview; if (filterStuck && !checkable.length && !humanReview.length) return null; // nothing to do — stay silent if (now - this.lastReviewTickAt < this.cfg.quietPeriodMs) return null; this.lastReviewTickAt = now; const aiKeys = checkable.map((i) => i.key); const humanKeys = humanReview.map((i) => i.key); const message = why === "stuck" ? `[orch-tick: review] AI-review in flight: ${aiKeys.join(", ") || "none"} — check each: if its reviewer finished, read the verdict (human-review / re-dispatch); if the reviewer is still running, leave it (its completion will notify); if it died, re-dispatch. Awaiting YOUR approval: ${humanKeys.join(", ") || "none"} — queue_update status: done to accept, active to re-dispatch — findings REQUIRED (queue_update(key, {status:'active', findings:''}); the harness's human re-dispatch sweep then spawns the redo — do NOT call queue_dispatch, it refuses active-status items). Not a user request; respond ≤2 lines.` : `[orch-tick: review] A reviewer completed. AI-review items: ${aiKeys.join(", ") || "none"}. Awaiting YOUR approval (human-review): ${humanKeys.join(", ") || "none"} — queue_update status: done to accept / active to re-dispatch (findings REQUIRED: queue_update(key, {status:'active', findings:''}); the sweep spawns the redo — do NOT call queue_dispatch) / rejected to drop. Not a user request; respond ≤2 lines.`; const suppressed = filterStuck ? aiReview.filter((i) => !checkable.includes(i)).map((i) => i.key) : []; return { reason: "review", message, facts: { aiReview: aiKeys, humanReview: humanKeys, count: aiKeys.length + humanKeys.length, why, ...(suppressed.length ? { suppressedInFlight: suppressed } : {}) }, }; } /** * Budget warning tick — ONE proactive per-run nudge at the timer heartbeat * when an ACTIVE budgeted run has consumed ~75%+ of its recorded wall-clock * budget. Warned keys are remembered for the Autopilot lifetime and pruned * once the run leaves the at-risk set, so a re-dispatch that runs long again * warns once more, but the same run is never nagged every 10 minutes. */ private budgetWarningTick(budget: QueueState["budget"]): Tick | null { const atRisk = budget.filter((b) => b.used >= 0.75); const riskKeys = new Set(atRisk.map((b) => b.key)); // Prune runs that finished (or were re-dispatched below the line). for (const k of this.budgetWarned) { if (!riskKeys.has(k)) this.budgetWarned.delete(k); } const fresh = atRisk.filter((b) => !this.budgetWarned.has(b.key)); if (!fresh.length) return null; for (const b of fresh) this.budgetWarned.add(b.key); const pctOf = (used: number): string => (used >= 1 ? "100%+" : `${Math.floor(used * 100)}%`); return { reason: "budget", message: `[orch-tick: budget] ${fresh.map((b) => `${b.key} has used ~${pctOf(b.used)} of its ${formatDurationMs(b.cap)} wall-clock budget (~${formatDurationMs(b.remaining)} left)`).join("; ")} — a budget cap will cut the run off mid-task. Steer it to wrap up + commit, or plan the bigger-budget re-dispatch now. Not a user request; respond ≤2 lines.`, facts: { budget: fresh.map((b) => ({ key: b.key, cap: b.cap, remaining: b.remaining, elapsed: b.elapsed, used: b.used })) }, }; } /** The OLDEST pending proposal's age in ms (null when none / unparseable). */ private oldestProposalAgeMs(now: number): number | null { const store = loadStore(this.cfg.stateDir); if (!store) return null; let oldest: number | null = null; for (const it of Object.values(store.items)) { if (it.status !== "proposal") continue; const t = Date.parse(it.createdAt || it.updatedAt || ""); if (Number.isNaN(t)) continue; if (oldest === null || t < oldest) oldest = t; } return oldest === null ? null : Math.max(0, now - oldest); } /** The key of the OLDEST pending proposal (for the intake tick's note). */ private oldestProposalKey(): string | null { const store = loadStore(this.cfg.stateDir); if (!store) return null; let oldestKey: string | null = null; let oldestT = Number.POSITIVE_INFINITY; for (const it of Object.values(store.items)) { if (it.status !== "proposal") continue; const t = Date.parse(it.createdAt || it.updatedAt || ""); if (Number.isNaN(t)) continue; if (t < oldestT) { oldestT = t; oldestKey = it.key; } } return oldestKey; } /** Public status snapshot for `/autopilot status`. * * `running` is the SAME derived occupancy every other reader uses * (liveTrackedSlots) — NEVER the engine's run ledger. The ledger is a * separately-incremented counter that only decrements on a completion event * with a matching id, so a stopped/killed run or a lost completion made it * ratchet up and never recover; showing it here was the same lie the tools * and the tick were fixed to stop telling. The host passes its backend so a * live run's own status.json is the liveness source, exactly as in the * tools; with no backend the no-backend grace fallback applies. */ status(backend?: Pick | null, now = this.now()): { running: number; lastTickAt: number; lastTickHash: string } { const store = loadStore(this.cfg.stateDir); const running = store ? liveTrackedSlots(store, { backend: backend ?? null, zombieGraceMinutes: this.cfg.zombieGraceMinutes, now }) : 0; return { running, lastTickAt: this.lastTickAt, lastTickHash: this.lastTickHash }; } // -- internals ------------------------------------------------------------ private readQueueSnapshot(now = this.now(), backend?: Pick | null): QueueState { const store = loadStore(this.cfg.stateDir); // No md fallback: state.md is retired (migration ran once at activation). // Missing store → empty snapshot → fail-safe to action (spurious tick costs // one turn; a missed one costs an idle slot). return store ? storeToSnapshot(store, now, { backend, zombieGraceMinutes: this.cfg.zombieGraceMinutes }) : emptyState(); } private decideTick( state: QueueState, source: string, runId: string | undefined, now: number, slotFreed: boolean, fleetTotalActive?: number, ): Tick | null { if (now - this.lastTickAt < this.cfg.quietPeriodMs) { // Within the quiet window, never repeat a tick with the same queue hash // (a completion re-arms it via the ledger-size hash component). if (this.lastTickHash === this.queueHash(state)) return null; } const tick = this.currentTick(state, source, runId, now, slotFreed, fleetTotalActive); if (tick) this.rememberTick(state, now); return tick; } /** The CURRENT-state tick decision — the single source of dispatch/intake * message text. decideTick calls it at GENERATION (behind the quiet/hash * gates above); refreshTickFacts calls it at DELIVERY so a busy-deferral * flush re-derives FLEET/QUEUE facts instead of presenting the snapshot * they were baked with. */ private currentTick( state: QueueState, source: string, runId: string | undefined, now: number, slotFreed: boolean, fleetTotalActive?: number, ): Tick | null { const slotsFree = Math.max(0, this.cfg.maxSlots - state.occupied); const ready = state.ready; const readyKeys = state.approved.map((a) => a.key); // approved = dispatchable (the fold) // DEPENDENCY GATE (KEY: AUTOPILOT-81): approved-but-unmet-deps items, named // so a full-but-waiting pipeline reads as "waiting on X", not as empty. const waitingKeys = state.waiting.map((w) => w.key); const waitingText = state.waiting.length ? `, ${state.waiting.length} waiting (after: ${state.waiting.map((w) => `${w.key}\u2192${w.after.join(",")}`).join("; ")})` : ""; // BUDGET-GOVERNANCE telemetry: any ACTIVE run past ~75% of its recorded // wall-clock budget is cap-risk — name it in the dispatch message so the // operator can steer/commit (or plan a bigger-budget re-dispatch) before // the cap cuts the run off. Facts always carry the full per-run // remaining/cap rows + the failed-at-cap recovery keys. const atRisk = state.budget.filter((b) => b.used >= 0.75); const pctOf = (used: number): string => (used >= 1 ? "100%+" : `${Math.floor(used * 100)}%`); const budgetLine = atRisk.length ? ` BUDGET: ${atRisk.map((b) => `${b.key} at ~${pctOf(b.used)} of its ${formatDurationMs(b.cap)} cap (~${formatDurationMs(b.remaining)} left)`).join("; ")} — cap risk: steer it to wrap up + commit, or plan a bigger-budget re-dispatch.` : ""; // Cross-check: when the authoritative fleet sees runs beyond what the // engine's own view (ledger + store — the `tracked` inventory) attributes, // flag it so the orchestrator can run `subagent status` on the untracked // ones (scouts, plain subagent calls). Occupied itself is the conservative // union — never a queue-only or an RPC-only number. const tracked = state.tracked ?? state.occupied; const reconcile = typeof fleetTotalActive === "number" && fleetTotalActive > tracked ? ` (fleet status shows ${fleetTotalActive} active — ${fleetTotalActive - tracked} not tracked by the queue)` : ""; // Capacity gap → dispatch tick (the core fix: refill on slot-free). // FLEET counts ALL session subagents (workers + reviewers + scouts — any // dispatched run occupies a slot); QUEUE comes from the store. if (slotFreed && slotsFree > 0 && ready >= 1) { return { reason: "dispatch", message: `[orch-tick: dispatch] FLEET: ${state.occupied}/${this.cfg.maxSlots} subagent runs active (workers + reviewers + scouts), ${slotsFree} free. ` + `QUEUE: ${ready} ready (${readyKeys.join(", ") || "none"})${waitingText}.` + reconcile + budgetLine + ` Rule: a slot is free + queue has ready work → dispatch it. System ping: run your loop. Respond ≤2 lines. Not a user request.`, facts: { slotsFree, occupied: state.occupied, ready, readyKeys, ...(state.waiting.length ? { waiting: state.waiting.length, waitingKeys } : {}), source, runId, fleetTotalActive, generatedAt: now, // Budget health: per-active-run remaining budget + cap (ms) and the // failed-at-cap recovery keys (failed=cap vs failed=verdict is the // failCause flag on each failed item). ...(state.budget.length ? { budget: state.budget.map((b) => ({ key: b.key, cap: b.cap, remaining: b.remaining, elapsed: b.elapsed, used: b.used })) } : {}), ...(state.budgetCappedFailed.length ? { budgetCapped: state.budgetCappedFailed } : {}), }, }; } // Queue drained below the buffer → intake tick (refill the approval buffer). // The ≤2-line rule does NOT apply here — intake ticks demand a REAL scan. // DEPENDENCY GATE (KEY: AUTOPILOT-81): the low-buffer check counts ready // + waiting together — a pipeline that is FULL but stuck on dependencies // must not also be told to run an intake scan for more work. if (ready + state.waiting.length < this.cfg.queueLowThreshold) { // INTAKE SUPPRESSION: while proposals from the previous intake are still // pending (the user is reading/thinking/discussing them), do NOT re-nudge // — ADDING proposals changes the queue hash, which would otherwise re-fire // this tick while the approved buffer is still low. The intake re-arms // when the proposals resolve (approved/rejected) or the queue changes. // AGE-AWARE: a proposal that has been pending for longer than the // suppression window (default 24h) must NOT starve the refill nudge // forever — the intake tick returns and names the stale proposal. if (state.proposalsPending > 0) { const suppressionMs = (this.cfg.intakeSuppressionHours ?? 24) * 3600 * 1000; const oldest = this.oldestProposalAgeMs(now); if (oldest !== null && oldest < suppressionMs) return null; } const stale = this.oldestProposalKey(); return { reason: "intake", message: `[orch-tick: intake] QUEUE: approved buffer low (${ready} ready < ${this.cfg.queueLowThreshold}; ready: ${readyKeys.join(", ") || "none"})${waitingText}. ` + `FLEET: not involved — this is about refilling the approved queue, not dispatch. ` + `Run a FULL intake scan NOW: scan ALL sources in order, diff against the queue, AND run the beyond-source pass (cross-source gaps, industry standards, project understanding). ` + `Propose the next batch for approval per the approval gate (scope + cwd + evidence + value/urgency + risk — ALL required, ENFORCED by the item contract). ` + `This is a real scan, not a quick check — present candidates. Not a user request; the ≤2-line rule does NOT apply to intake ticks.` + (stale ? ` NOTE: ${stale} has been pending for a while — resolve or reject it so it stops blocking the buffer.` : ""), facts: { ready, readyKeys, ...(state.waiting.length ? { waiting: state.waiting.length, waitingKeys } : {}), threshold: this.cfg.queueLowThreshold, source, runId, generatedAt: now }, }; } return null; } /** * DELIVERY-TIME FACT REFRESH (AUTOPILOT-6). A dispatch/intake tick that the * busy-deferral held is flushed at the host's NEXT settle — possibly long * after generation, when the store has moved underneath it (items the tick * listed as ready were dispatched mid-window, completions flipped items, * auto-dispatch filled slots). Re-derive the facts from the LIVE store (+ * authoritative fleet when the caller has one, else the conservative * ledger/store fallback) so a late-delivered tick NEVER claims "X free, * N ready (…)" while those items are running: * - returns the CURRENT truthful nudge (dispatch-first, then intake) with * fresh facts + a refreshedAt stamp (generatedAt preserved for the audit * trail); null when NO nudge applies to the current state — the flush * drops the stale message rather than deliver it; * - review ticks pass through untouched (their facts are event-scoped — * key/verdict — they carry no FLEET/QUEUE claims). * The CURRENT state's hash is remembered so the same-settle sweep does not * re-fire the identical fresh nudge. */ refreshTickFacts(tick: Tick, fleet?: { totalActive?: number; occupied?: number }, now = this.now(), backend?: Pick | null): Tick | null { if (tick.reason !== "dispatch" && tick.reason !== "intake") return tick; const snapshot = this.readQueueSnapshot(now, backend); // Same FLEET-vs-INVENTORY PARITY as sweep: the authoritative fleet is the // floor; the delivery-time numbers are the store's own occupied count. // KEY: AUTOPILOT-71 — no ledger term here either, same fix/tradeoff as // sweep() above (see its comment): a leaked ledger entry no longer // inflates this reading, but a scout/ad-hoc run the store never recorded // is now invisible to the refreshed facts too, unless the fleet RPC // catches it. // // SAME one-occupancy rule as sweep() (KEY: AUTOPILOT-96): a refresh of a // held dispatch/intake tick projects the gate's occupancy (passed as // `occupied`) unioned with the live store, never the store-live-only view // that used to let it claim a free slot the gate did not have. const tracked = snapshot.occupied; const occupied = Math.max(fleet?.occupied ?? 0, fleet?.totalActive ?? 0, tracked); const eff = { ...snapshot, occupied, tracked }; const source = typeof tick.facts.source === "string" ? tick.facts.source : "settled"; const fresh = this.currentTick(eff, source, undefined, now, true, fleet?.totalActive); if (!fresh) return null; // the nudge no longer applies — the stale message must not be delivered this.rememberTick(eff, now); // arm the same-settle sweep dedupe with the CURRENT hash const generatedAt = typeof tick.facts.generatedAt === "number" ? tick.facts.generatedAt : now; const facts = { ...fresh.facts, generatedAt, refreshedAt: now }; // Delivery-time telemetry: the gap between the generation `tick` line and // this `tick-refresh` line IS the busy-deferral delay — and `changed` // records whether the live facts moved underneath the held message. this.logEvent("tick-refresh", { reason: fresh.reason, from: tick.reason, changed: fresh.message !== tick.message }); return { ...fresh, facts }; } private queueHash(state: QueueState): string { // Include the fleet count so a completion (ledger shrink) re-arms the // settle check even when the queue content itself did not change. return stateHash(state) + `|fleet:${this.ledger.size}`; } private rememberTick(state: QueueState, now: number): void { this.lastTickAt = now; this.lastTickHash = this.queueHash(state); } private now(): number { return this.cfg.now ? this.cfg.now() : Date.now(); } private logEvent(type: string, data: Record): void { if (!this.cfg.log) return; try { this.cfg.log(JSON.stringify({ t: new Date().toISOString(), type, ...data })); } catch { // silent } } } // --------------------------------------------------------------------------- // Snapshot helpers // --------------------------------------------------------------------------- /** Derive the queue facts the tick engine needs from the store (deterministic). * Occupied = the ONE live-item derivation (liveTrackedSlots): `active` items * (workers) plus `ai-review` items with a LIVE reviewer — a run whose own * record says stopped/killed/terminal holds no slot, so a stopped run releases * it with no completion event. This is the store half of the fleet number — * the fleet RPC totalActive is unioned on top by sweep()/refreshTickFacts() * and covers runs the queue doesn't track (scouts, plain subagent calls). * `now` also drives the budget telemetry: elapsed = now − the item's * flip-to-active updatedAt stamp (≈ dispatch time), remaining = cap − elapsed. */ export function storeToSnapshot(store: QueueStore, now = Date.now(), opts: Pick = {}): QueueState { const items = Object.values(store.items); const active = items .filter((i) => i.status === "active") .map((i) => ({ key: i.key, runId: i.runId ?? undefined, status: "working" as const, title: i.title, line: "", lineIndex: 0, section: "active" as const })); const budget = active .filter((a) => { const it = store.items[a.key]; return it.timeoutMs !== null && it.timeoutMs !== undefined && it.timeoutMs > 0; }) .map((a) => { const it = store.items[a.key]; const cap = it.timeoutMs as number; const startedAt = Date.parse(it.updatedAt || ""); const elapsed = Number.isFinite(startedAt) ? Math.max(0, now - startedAt) : 0; return { key: a.key, cap, elapsed, remaining: Math.max(0, cap - elapsed), used: cap > 0 ? elapsed / cap : 0 }; }); const budgetCappedFailed = items .filter((i) => i.status === "failed" && i.failCause === "budget-capped") .map((i) => i.key); // DEPENDENCY GATE (KEY: AUTOPILOT-81): an `approved` item with an unmet // `after` is WAITING, not ready — split out of `approved`/`ready` here so // every consumer of the snapshot (tick text, /autopilot status via // queue_list) sees the split for free. const approvedItems = items.filter((i) => i.status === "approved"); const waitingItems = approvedItems.filter((i) => unmetDependencies(i, store).length > 0); const waitingKeys = new Set(waitingItems.map((i) => i.key)); const approved = approvedItems .filter((i) => !waitingKeys.has(i.key)) .map((i) => ({ key: i.key, title: i.title, line: "", lineIndex: 0 })); const waiting = waitingItems.map((i) => ({ key: i.key, title: i.title, after: unmetDependencies(i, store) })); const occupied = liveTrackedSlots(store, { ...opts, now }); const ready = approved.length; // approved = dispatchable now (unmet-dependency items excluded) const proposalsPending = items.filter((i) => i.status === "proposal").length; return { active, approved, waiting, occupied, ready, proposalsPending, ok: true, budget, budgetCappedFailed }; } function stateHash(state: QueueState): string { const active = state.active .map((a) => `${a.key}:${a.runId ?? ""}:${a.status ?? ""}`) .sort() .join("|"); const approved = state.approved .map((a) => a.key) .sort() .join("|"); // DEPENDENCY GATE (KEY: AUTOPILOT-81): the waiting SET is part of the hash // too — an item entering/leaving `waiting` (a dependency finishing, a new // waiting item appearing) must re-arm the settle check exactly like an // approved/active membership change does. const waiting = state.waiting .map((w) => w.key) .sort() .join("|"); return `${state.occupied}|${state.ready}|${state.proposalsPending}|${active}|${approved}|${waiting}`; } function emptyState(): QueueState { return { active: [], approved: [], waiting: [], occupied: 0, ready: 0, proposalsPending: 0, ok: false, budget: [], budgetCappedFailed: [] }; } function empty(): AutopilotResult { return { tick: null, domainEvents: [], flipped: false, freedSlot: false, reviewerCompleted: false }; } /** Classify a run condemned by deadRunReconcile from its status.json `error`. * The runtime spells a budget kill "Subagent timed out after ms" — routing * that to `budget-capped` (not `zombie`) is what makes auto-recovery re-dispatch * it as a CAP instead of a verdict. Anything else is a lost completion event, * which is exactly what `zombie` means. */ function deadRunFailCause(error: string | null): FailCause { return error && /timed out after \d+\s*ms/i.test(error) ? "budget-capped" : "zombie"; } /** Classify a FAILED worker run: a budget CAP cut it off (runtime reported * timedOut, or the run lived at/over the item's recorded timeoutMs — the * deterministic backstop when the runtime does not say 'timed out'), versus * an unsuccessful verdict/exit. Elapsed comes from the item's updatedAt (the * flip-to-active stamp ≈ dispatch time); only items with a recorded budget * are ever classified by elapsed. * * AUTOPILOT-53: `timeoutMs`, when recorded, IS what the runtime is honouring * — there is no ceiling to compare against instead, so the backstop compares * elapsed directly against the recorded budget. An item with NO recorded * budget makes no promise the backstop can check, so it is left to * `ev.timedOut` alone (never defaulted to the runtime's internal fallback * here — that would misclassify a merely long-running verdict/zombie * failure as a cap on every unbudgeted item, which is exactly the false * positive this comment warns against). */ function workerFailCause(ev: CompletionEvent, timeoutMs: number | null | undefined, updatedAt: string, now: number): FailCause { if (ev.timedOut === true) return "budget-capped"; const honouredMs = typeof timeoutMs === "number" && Number.isFinite(timeoutMs) && timeoutMs > 0 ? timeoutMs : null; if (honouredMs !== null) { const startedAt = Date.parse(updatedAt || ""); if (Number.isFinite(startedAt) && now - startedAt >= honouredMs) return "budget-capped"; } // PROVIDER/INFRA (AUTO-RECOVER-FAILS): a run that ended unsuccessful with NO // deliverable text at all — the provider returned an empty api_error / the // session produced nothing. That is not a verdict on the work, so it gets the // bounded provider-retry recovery (2× then escalate). Detected only when the // event carries a results array whose entries have no output AND no summary; // an event with no results at all stays a verdict (legacy/unenriched runs). if (infraNoOutput(ev)) return "spawn"; return "verdict"; } /** true when the completion carries results but produced NO output text. */ function infraNoOutput(ev: CompletionEvent): boolean { if (typeof ev.summary === "string" && ev.summary.trim()) return false; if (!Array.isArray(ev.results) || ev.results.length === 0) return false; return ev.results.every((r) => { const out = (r as { output?: unknown })?.output; return !(typeof out === "string" && out.trim()); }); }