From dc25112d20fb68af9a5898a979229d68fb419459 Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Sat, 10 Oct 2026 00:16:57 -0400 Subject: [PATCH 1/2] feat(subagent): require replay safety for recovery availability A worker that wrote files or ran a shell command before its provider gave out was still marked continuable, so a successor with the same brief could repeat those side effects. Recovery is now available only when every tool the run requested or started is a read, search, or web read; anything else, including MCP and unknown tools, fails closed to side_effects_completed and normal director-led continuation. --- src/agent/tool-classification.ts | 13 +++++ .../session-store-failure-record.test.ts | 26 +++++++++ src/subagent/session-store.ts | 5 ++ src/subagent/terminal-failure.test.ts | 53 ++++++++++++++++++- src/subagent/terminal-failure.ts | 13 +++++ 5 files changed, 108 insertions(+), 2 deletions(-) diff --git a/src/agent/tool-classification.ts b/src/agent/tool-classification.ts index 8af13343e..09c5c77f7 100644 --- a/src/agent/tool-classification.ts +++ b/src/agent/tool-classification.ts @@ -29,6 +29,19 @@ export const SEARCH_QUERY_TOOLS: ReadonlySet = new Set([ "search_files", ]); +/** + * Tools whose call a recovery may replay without repeating a side effect: + * path-keyed reads, search queries, and web reads. Deliberately narrow and + * closed: run_shell, writes, MCP, and any tool not named here make a failed + * run ineligible for enforced recovery. + */ +export const REPLAY_SAFE_TOOLS: ReadonlySet = new Set([ + ...PATH_KEYED_READ_TOOLS, + ...SEARCH_QUERY_TOOLS, + "web_fetch", + "web_search", +]); + /** * Tools that never need an approval prompt (cannot change the workspace): * the director's read surface minus run_shell/web_fetch/web_search, which diff --git a/src/subagent/session-store-failure-record.test.ts b/src/subagent/session-store-failure-record.test.ts index 6a10edc35..a84293dac 100644 --- a/src/subagent/session-store-failure-record.test.ts +++ b/src/subagent/session-store-failure-record.test.ts @@ -186,6 +186,32 @@ describe("terminal failure record", () => { expect(after?.error).toBe("first"); }); + test("a retryable failure after a write or shell call is not replay-safe", () => { + for (const tool of ["write_file", "run_shell", "mcp__x__y"]) { + const { sessions } = store(); + const id = startRunning(sessions); + sessions.appendEvent(id, toolStart("read_file", "c1")); + sessions.appendEvent(id, toolStart(tool, "c2")); + sessions.fail(id, "timeout", { failure_class: "provider_retryable" }); + expect(sessions.get(id)?.failure?.recovery).toEqual({ + available: false, + reason: "side_effects_completed", + }); + } + }); + + test("a retryable failure after only read calls stays eligible", () => { + const { sessions } = store(); + const id = startRunning(sessions); + sessions.appendEvent(id, toolStart("read_file", "c1")); + sessions.appendEvent(id, toolStart("grep", "c2")); + sessions.fail(id, "timeout", { failure_class: "provider_retryable" }); + expect(sessions.get(id)?.failure?.recovery).toEqual({ + available: true, + reason: "eligible", + }); + }); + test("a teardown failure reports partial cleanup", () => { const { sessions } = store(); const id = startRunning(sessions); diff --git a/src/subagent/session-store.ts b/src/subagent/session-store.ts index 4f386b3e8..5062f7ba7 100644 --- a/src/subagent/session-store.ts +++ b/src/subagent/session-store.ts @@ -33,6 +33,7 @@ import { canonicalToolName } from "../agent/canonical-tool-name.js"; import { classifySubAgentFailure, failureClassForStopReason, + isReplaySafeToolName, isSubAgentLoopGuardError, recoveryFor, type SubAgentFailureClass, @@ -843,6 +844,10 @@ export function createSubAgentSessionStore( attempt, followupTurn: followupTurns.has(id), cause: pendingFailures.get(id)?.cause, + // Every tool the run requested, started, or finished: a call + // that started may have landed its side effect before the run + // died, so this is a superset of completed calls. + replaySafe: session.toolNames.every(isReplaySafeToolName), }), ), attempt, diff --git a/src/subagent/terminal-failure.test.ts b/src/subagent/terminal-failure.test.ts index ba4c8f6e9..68ba96fb4 100644 --- a/src/subagent/terminal-failure.test.ts +++ b/src/subagent/terminal-failure.test.ts @@ -6,6 +6,7 @@ import { createSubAgentLoopGuardError, failureClassForStopReason, isProviderFailureClass, + isReplaySafeToolName, isSubAgentLoopGuardError, recoveryFor, } from "./terminal-failure.js"; @@ -108,6 +109,7 @@ describe("recoveryFor", () => { failureClass: "provider_retryable", attempt: 1, followupTurn: false, + replaySafe: true, }), ).toEqual({ available: true, reason: "eligible" }); }); @@ -120,7 +122,12 @@ describe("recoveryFor", () => { "error", ] as const)("%s is not_retryable", (failureClass) => { expect( - recoveryFor({ failureClass, attempt: 1, followupTurn: false }), + recoveryFor({ + failureClass, + attempt: 1, + followupTurn: false, + replaySafe: true, + }), ).toEqual({ available: false, reason: "not_retryable" }); }); @@ -131,6 +138,7 @@ describe("recoveryFor", () => { attempt: 1, followupTurn: false, cause: providerError("retryable"), + replaySafe: true, }), ).toEqual({ available: true, reason: "eligible" }); expect( @@ -139,6 +147,7 @@ describe("recoveryFor", () => { attempt: 1, followupTurn: false, cause: createSubAgentLoopGuardError(providerError("timeout")), + replaySafe: true, }), ).toEqual({ available: true, reason: "eligible" }); }); @@ -150,6 +159,7 @@ describe("recoveryFor", () => { attempt: 1, followupTurn: false, cause: providerError("credential_failure"), + replaySafe: true, }), ).toEqual({ available: false, reason: "not_retryable" }); expect( @@ -158,6 +168,7 @@ describe("recoveryFor", () => { attempt: 1, followupTurn: false, cause: new Error("doom"), + replaySafe: true, }), ).toEqual({ available: false, reason: "not_retryable" }); expect( @@ -166,6 +177,7 @@ describe("recoveryFor", () => { attempt: 1, followupTurn: true, cause: providerError("retryable"), + replaySafe: true, }), ).toEqual({ available: false, reason: "not_retryable" }); expect( @@ -174,6 +186,7 @@ describe("recoveryFor", () => { attempt: 2, followupTurn: false, cause: providerError("retryable"), + replaySafe: true, }), ).toEqual({ available: false, reason: "recovery_exhausted" }); }); @@ -184,14 +197,50 @@ describe("recoveryFor", () => { failureClass: "provider_retryable", attempt: 1, followupTurn: true, + replaySafe: true, }), ).toEqual({ available: false, reason: "not_retryable" }); }); + test("a retryable failure after a non-replay-safe call has side effects", () => { + expect( + recoveryFor({ + failureClass: "provider_retryable", + attempt: 1, + followupTurn: false, + replaySafe: false, + }), + ).toEqual({ available: false, reason: "side_effects_completed" }); + }); + + test("only read, search, and web read tools are replay-safe", () => { + for (const name of ["read_file", "read", "grep", "search_files"]) { + expect(isReplaySafeToolName(name)).toBe(true); + } + for (const name of ["web_fetch", "web_search"]) { + expect(isReplaySafeToolName(name)).toBe(true); + } + for (const name of [ + "write_file", + "edit_file", + "run_shell", + "bash", + "mcp__server__tool", + "some_unknown_tool", + ]) { + expect(isReplaySafeToolName(name)).toBe(false); + } + }); + test("a failed recovery attempt is exhausted whatever its class", () => { for (const failureClass of ["provider_retryable", "error"] as const) { expect( - recoveryFor({ failureClass, attempt: 2, followupTurn: false }), + recoveryFor({ + failureClass, + attempt: 2, + followupTurn: false, + replaySafe: true, + }), ).toEqual({ available: false, reason: "recovery_exhausted" }); } }); diff --git a/src/subagent/terminal-failure.ts b/src/subagent/terminal-failure.ts index 7aba584af..0e4986393 100644 --- a/src/subagent/terminal-failure.ts +++ b/src/subagent/terminal-failure.ts @@ -3,6 +3,8 @@ import { isResolvedProviderFailureError, } from "../inference-error-message.js"; import { errorMessage } from "../agent/error-message.js"; +import { canonicalToolName } from "../agent/canonical-tool-name.js"; +import { REPLAY_SAFE_TOOLS } from "../agent/tool-classification.js"; import type { ForcedStopReason } from "./stop-policy.js"; /** @@ -125,15 +127,23 @@ export function classifySubAgentFailure( return "error"; } +/** Whether a call to `name` can be replayed by a recovery attempt. */ +export function isReplaySafeToolName(name: string): boolean { + return REPLAY_SAFE_TOOLS.has(canonicalToolName(name)); +} + /** * Recovery availability for a freshly written record. The chain cap wins * over class: a recovery attempt that fails is exhausted whatever its cause. + * Replay safety fails closed: any tool outside the read-only allowlist means + * a replacement could repeat a side effect. */ export function recoveryFor(input: { failureClass: SubAgentFailureClass; attempt: number; followupTurn: boolean; cause?: unknown; + replaySafe: boolean; }): SubAgentRecovery { if (input.attempt > 1) { return { available: false, reason: "recovery_exhausted" }; @@ -141,6 +151,9 @@ export function recoveryFor(input: { if (input.followupTurn || !isRetryableFailure(input)) { return { available: false, reason: "not_retryable" }; } + if (!input.replaySafe) { + return { available: false, reason: "side_effects_completed" }; + } return { available: true, reason: "eligible" }; } From 782894e3875f535e6881f0e8fbf5e8d1b887b8c8 Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Sat, 10 Oct 2026 00:17:26 -0400 Subject: [PATCH 2/2] feat(subagent): enforce one recovery per failure via spawn_agent The single-successor hint for a transient worker failure was advisory: nothing tied a successor to the failed run, counted it, or stopped a second one. spawn_agent now takes recovers, the failed agent id. The claim is recorded synchronously before the handler awaits, so duplicate or concurrent calls get the same replacement back and spawn nothing; an ineligible request is refused with its typed reason and never falls back to a plain spawn. A replacement is attempt 2, so its own failure is recovery_exhausted, and it is an ordinary spawn that inherits no grants, approvals, or pending asks. Reverting this restores the advisory guidance. --- docs/ARCHITECTURE.md | 2 +- src/agent/prompts.ts | 2 +- src/subagent/agent-fleet.ts | 137 +++- src/subagent/fleet-dry-drive.ts | 13 +- src/subagent/fleet-recovery.test.ts | 631 ++++++++++++++++++ .../session-store-failure-record.test.ts | 35 + src/subagent/session-store.ts | 44 +- src/subagent/terminal-failure.ts | 32 + 8 files changed, 884 insertions(+), 12 deletions(-) create mode 100644 src/subagent/fleet-recovery.test.ts diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 21c96d088..d705a8b6c 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -153,7 +153,7 @@ Two directors, selected by role: Auto mode is toggled by CLI flags (`--auto` / `--no-auto`); there is currently no in-session key to toggle it (default on; constrained envelope — workspace writes and unconstrained shell auto-allow; installs, recursive rm, force/uncontained worktree changes, sensitive-path and opaque-wrapper shell still ask; contained non-force `git worktree add`/`remove`/`prune` and `list` auto-allow; shell file-mutation denied). It is not a separate edit/plan mode. - **SubAgentDirector** (delegated work, `src/subagent/index.ts`) — Drives a dispatched worker until a turn arrives with no tool calls, then replies with the final assistant text and ends the run. A tool-less turn **after tools** completes only with the four-heading envelope (Summary, Findings, Blockers, Paths). Assistant text that prints explicit `` markup is treated as attempted tool use, not narration: one **verbatim-tool-call** nudge asks the worker to re-issue a real `tool_call` and does not count toward the tool-less spiral. A missing envelope otherwise nudges once (**incomplete-report**) and a second tool-less turn still without the envelope salvages as **incomplete-report-stop** (once; later tool-less turns wait). Explore/read-only workers that used tools then replied with findings remain normal completes; `requireEvidence` (off by default, set per director) additionally requires at least one read before a tool-less spawn-only reply can complete. `requirePlanSubstance` (planner or `intent=plan`, not `modelRole === "plan"`) additionally requires Findings to contain files/paths, acceptance criteria, non-goals, risks, and ordered steps with a non-placeholder line each — four headings with stub Findings are incomplete-report, not an attachable plan. After real tool work, wrap-up Findings that are not placeholder or outline-only complete instead of being salvaged as a stub plan. Reads done through `run_shell` count as evidence too — `src/subagent/shell-evidence.ts` classifies shell reads (`cat`, `grep`, `sed` without `-i`, …) over the same subject expansion the auto-shell policy uses — but there is no corresponding shell-write evidence or file-write requirement: a run that never touches a file still completes normally once it replies with the envelope. There is no turn budget. Operator/parent cancel after any progress returns a **cancelled** salvage report (partial findings + tool activity) instead of a bare cancel string; cancel before progress still surfaces as cancelled-by-operator. There is no repetition/no-progress/never-acted/never-edited hard stop and no fingerprint-based re-dispatch block — a genuinely stuck worker runs until it completes, stalls, hits an opt-in wall-clock deadline, or is cancelled. - `spawn_agent` starts each worker and records it in the caller's fleet mailbox. On the TUI primary, mailbox mail is the collect path: occupancy takes uncollected terminals and re-enters the parent as system inbound with a Summary + Blockers digest (Findings when Summary is empty) and a blob pointer named for `read_file`. Nested orchestrators collect through mailbox mail the same way. A mounted `wait_agents` (exec primary) collects worker reports directly instead. Already-collected waits return status without a second report or error body. Wait JSON includes `stop_reason` from the session when present so a salvage that is wait-`done` is not mistaken for a clean complete, and so parent-initiated interrupt (`interrupted`) is not mistaken for operator-cancel (`cancelled`). Every terminal non-clean outcome (failed, cancelled, interrupted, or a forced-stop salvage) also carries a `failure` record, written in the same store mutation that makes the outcome visible and never rewritten: a typed `failure_class` (`provider_retryable`, `provider_fatal`, `loop_guard` for the reactor's doom-loop guard, the forced-stop reasons, or `error`), `recovery` availability with a typed reason, the attempt number, the ask_director `handoff` state at termination (`delivered` only when the worker's own ask_director call returned the answer), and whether `cleanup` released everything. `provider_failure` and `continuable` derive from that record, so wait_agents, list_agents, and mailbox mail cannot disagree. Deadline salvage prepends an advisory parent hint suggesting continuation plus a longer deadline if more wall-clock time is warranted. Failed and incomplete-report salvage tell the parent to diagnose from the report or error and MAY spawn one successor with a changed brief. A parent-initiated interrupt is a resumable pause: wait unblocks with `stop_reason: interrupted` (often while the session is still running and has no report); the parent should `resume_agent` or re-wait, and must not spawn a successor against a still-live worker. Successor only if that session is no longer resumable. Operator-cancelled salvage asks the parent to synthesize Findings and Paths and wait for the operator instead of auto-starting another specialist. Identical re-dispatch of the same brief stays refused at the prompt / spawn-handoff layer, except a recoverable/continuable child failure (`continuable: true`) MAY spawn one successor with the same brief; there is no fingerprint-based re-dispatch hard-block. Deadline hints are advisory only — an identical re-dispatch is still admitted at runtime. Parent hints are prepended on salvage reports returned to the parent. The runtime does not auto-spawn successors. + `spawn_agent` starts each worker and records it in the caller's fleet mailbox. On the TUI primary, mailbox mail is the collect path: occupancy takes uncollected terminals and re-enters the parent as system inbound with a Summary + Blockers digest (Findings when Summary is empty) and a blob pointer named for `read_file`. Nested orchestrators collect through mailbox mail the same way. A mounted `wait_agents` (exec primary) collects worker reports directly instead. Already-collected waits return status without a second report or error body. Wait JSON includes `stop_reason` from the session when present so a salvage that is wait-`done` is not mistaken for a clean complete, and so parent-initiated interrupt (`interrupted`) is not mistaken for operator-cancel (`cancelled`). Every terminal non-clean outcome (failed, cancelled, interrupted, or a forced-stop salvage) also carries a `failure` record, written in the same store mutation that makes the outcome visible and never rewritten: a typed `failure_class` (`provider_retryable`, `provider_fatal`, `loop_guard` for the reactor's doom-loop guard, the forced-stop reasons, or `error`), `recovery` availability with a typed reason, the attempt number, the ask_director `handoff` state at termination (`delivered` only when the worker's own ask_director call returned the answer), and whether `cleanup` released everything. `provider_failure` and `continuable` derive from that record, so wait_agents, list_agents, and mailbox mail cannot disagree. Deadline salvage prepends an advisory parent hint suggesting continuation plus a longer deadline if more wall-clock time is warranted. Failed and incomplete-report salvage tell the parent to diagnose from the report or error and MAY spawn one successor with a changed brief. A parent-initiated interrupt is a resumable pause: wait unblocks with `stop_reason: interrupted` (often while the session is still running and has no report); the parent should `resume_agent` or re-wait, and must not spawn a successor against a still-live worker. Successor only if that session is no longer resumable. Operator-cancelled salvage asks the parent to synthesize Findings and Paths and wait for the operator instead of auto-starting another specialist. Identical re-dispatch of the same brief stays refused at the prompt / spawn-handoff layer, except a recoverable/continuable child failure (`continuable: true`), which the parent recovers by calling `spawn_agent` with `recovers` set to the failed agent_id; there is no fingerprint-based re-dispatch hard-block. Recovery is enforced, not advisory: `recovery.available` requires a `provider_retryable` failure on the original spawn (not a resume_agent turn), no prior recovery, an original attempt (a failed replacement is `recovery_exhausted`), and a replay-safe run that touched only read, search, and web read tools (anything else is `side_effects_completed`). The claim lands synchronously before the handler awaits, so a duplicate or concurrent `recovers` call gets the existing replacement back and spawns nothing; an ineligible one is refused with its reason and never falls back to a plain spawn. The replacement is an ordinary spawn: it inherits no grants, approvals, or pending asks from the failed worker. Deadline hints are advisory only — an identical re-dispatch is still admitted at runtime. Parent hints are prepended on salvage reports returned to the parent. The runtime does not auto-spawn successors. #### Model-family policy (`src/agent/model-family-policy.ts`) diff --git a/src/agent/prompts.ts b/src/agent/prompts.ts index 17b1aa8e9..953508151 100644 --- a/src/agent/prompts.ts +++ b/src/agent/prompts.ts @@ -210,7 +210,7 @@ const GUIDELINE_SUB_BLOCKS: Record< (ctx.waitAgentsMounted ? " or re-wait" : ", or idle for its mailbox mail") + - "; do not `spawn_agent` a successor against a still-live worker. Successor only if that session is no longer resumable. Operator-cancel (`stop_reason` cancelled): wait for the operator; do not auto-retry. Identical brief: refuse. Recoverable/continuable child failure (`continuable: true`) MAY spawn one successor with the same brief; identical brief is still refused otherwise. Merge Summary/Findings into a coherent answer for the operator; do not paste raw fleet-agent dumps.", + "; do not `spawn_agent` a successor against a still-live worker. Successor only if that session is no longer resumable. Operator-cancel (`stop_reason` cancelled): wait for the operator; do not auto-retry. Identical brief: refuse. Recoverable/continuable child failure (`continuable: true`): recover it at most once with `spawn_agent` and `recovers` set to its agent_id; the runtime refuses any other recovery with a reason. Identical brief is still refused otherwise. Merge Summary/Findings into a coherent answer for the operator; do not paste raw fleet-agent dumps.", "- Use manage_tasks for your own coordination checklist; spawning workers is `spawn_agent`, not manage_tasks.", "- If context is compacted automatically, do not stop tasks early due to token fear; persist progress via manage_tasks and worker reports.", ]; diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index a3d07c94e..360dbbc55 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -114,7 +114,10 @@ import { classifySubAgentFailure, isProviderFailureClass, isSubAgentLoopGuardError, + withoutRecoveryReplacement, + withRecoveryReplacement, type SubAgentFailureRecord, + type SubAgentRecoveryReason, } from "./terminal-failure.js"; import { errorMessage } from "../agent/error-message.js"; import { isSubAgentCancelError } from "./dispose.js"; @@ -270,6 +273,45 @@ class FleetMailbox { this.sessions.wake(); } + /** + * Record that `replacementId` recovers `failedId`. Runs synchronously in + * the spawn handler before anything awaits, so a second claim in the same + * tick already sees the replacement. An evicted session's record lives + * only on the overlay, so it is linked there. + */ + claimRecovery(failedId: string, replacementId: string): void { + const overlay = this.records.get(failedId); + if (overlay === undefined) return; + if (this.sessions.get(failedId) !== undefined) { + this.sessions.linkRecoveryReplacement(failedId, replacementId); + this.mirrorFailure(failedId, overlay); + return; + } + if ( + overlay.failure !== undefined && + overlay.failure.recovery.replacement_id === undefined + ) { + overlay.failure = withRecoveryReplacement(overlay.failure, replacementId); + } + } + + /** + * Give back a grant whose replacement never launched. Keyed on the + * replacement id so only the claimant can refund its own claim. + */ + refundRecovery(failedId: string, replacementId: string): void { + const overlay = this.records.get(failedId); + if (overlay === undefined) return; + if (this.sessions.get(failedId) !== undefined) { + this.sessions.unlinkRecoveryReplacement(failedId, replacementId); + this.mirrorFailure(failedId, overlay); + return; + } + if (overlay.failure?.recovery.replacement_id === replacementId) { + overlay.failure = withoutRecoveryReplacement(overlay.failure); + } + } + markQueued(id: string): void { const existing = this.records.get(id); if (existing === undefined || existing.collected === true) return; @@ -544,6 +586,7 @@ const SpawnAgentArgs = type({ "do_not?": "string[]", "report_focus?": "string", "requires_tools?": "string[]", + "recovers?": "string", }); export const spawnAgentToolDefinition: ToolDefinition = { @@ -596,6 +639,11 @@ export const spawnAgentToolDefinition: ToolDefinition = { description: "Hard tool requirements (canonical names). Preflight fails closed when missing.", }, + recovers: { + type: "string", + description: + "agent_id of a failed+continuable worker this spawn replaces. Once per failure; refused with a reason otherwise.", + }, }, required: ["description", "prompt"], }, @@ -613,7 +661,7 @@ export const MAX_WAIT_TIMEOUT_MS = 300_000; export const waitAgentsToolDefinition: ToolDefinition = { name: "wait_agents", description: - "Block until targets finish, fail, or ask (awaiting_director), or timeout_ms. mode any (default) or all. Timeout is not an error; workers keep running. Answer awaiting_director with send_input. failed+continuable:true may be respawned once.", + "Block until targets finish, fail, or ask (awaiting_director), or timeout_ms. mode any (default) or all. Timeout is not an error; workers keep running. Answer awaiting_director with send_input. failed+continuable:true: recover once via spawn_agent recovers=agent_id.", inputSchema: { type: "object", properties: { @@ -896,6 +944,49 @@ function handoffRequiresSuccessCriteria( ); } +/** + * Why a `recovers` request spawns nothing, or undefined when it may proceed. + * A duplicate is not an error: it hands back the replacement already + * started so every caller converges on one agent_id. + */ +function recoveryRefusal( + callId: string, + fleetRecords: FleetMailboxHandle, + failedId: string, +): ToolResult | undefined { + const record = fleetRecords.peek(failedId); + if (record === undefined) { + return fleetResult( + callId, + `Error: spawn_agent recovers="${failedId}" is not one of your workers; nothing was spawned.`, + ); + } + const replacementId = record.failure?.recovery.replacement_id; + if (replacementId !== undefined) { + return fleetResult( + callId, + fleetJson({ + agent_id: replacementId, + recovers: failedId, + status: "recovery_already_started", + reason: "already_recovered" satisfies SubAgentRecoveryReason, + }), + ); + } + if ( + record.status === "failed" && + record.failure?.recovery.available === true + ) { + return undefined; + } + const reason: SubAgentRecoveryReason = + record.failure?.recovery.reason ?? "not_retryable"; + return fleetResult( + callId, + `Error: recovery_unavailable reason=${reason}: spawn_agent recovers="${failedId}" spawned nothing. Diagnose from the failure and continue with a changed brief, or wait for the operator.`, + ); +} + export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { const telemetry = deps.telemetry ?? NOOP_TELEMETRY; // Overlap detection (replaces the static per-package writePaths lock): @@ -966,6 +1057,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { do_not: rawDoNot, report_focus: rawReportFocus, requires_tools: rawRequiresTools, + recovers: rawRecovers, } = parsed; const description = rawDesc.trim(); const prompt = rawPrompt.trim(); @@ -975,6 +1067,17 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { "Error: spawn_agent requires a non-empty description and prompt.", ); } + // Some models fill every optional field, sending recovers: "" on a + // fresh spawn; blank means absent, as for context and report_focus. + const trimmedRecovers = rawRecovers?.trim(); + const recovers = + trimmedRecovers === undefined || trimmedRecovers.length === 0 + ? undefined + : trimmedRecovers; + if (recovers !== undefined) { + const refusal = recoveryRefusal(call.id, deps.fleetRecords, recovers); + if (refusal !== undefined) return refusal; + } const context = rawCtx?.trim(); const goals = rawGoals?.map((g) => g.trim()).filter((g) => g.length > 0) ?? []; @@ -1218,6 +1321,11 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { : {}), }); + // CL-10202: the claim lands before this handler's first await (and + // before the session exists), so a concurrent duplicate sees it. + if (recovers !== undefined) { + deps.fleetRecords.claimRecovery(recovers, call.id); + } const session = deps.sessions.start({ id: call.id, description, @@ -1231,6 +1339,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { ...(deps.parentSessionId !== undefined ? { parentSessionId: deps.parentSessionId } : {}), + ...(recovers !== undefined ? { recovers } : {}), }); deps.fleetRecords.register(session.id); const agentName = classifyAgentName(resolved.directorId); @@ -1273,9 +1382,21 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { }); }; + // The claim precedes launch so concurrent recoveries cannot both + // start, which means every exit that never reaches run() must hand the + // grant back or the failed lane is frozen on a replacement that does + // not exist. Once run() has been called the replacement really was + // attempted, so a failure from there on keeps the grant spent. + let runBegan = false; + const refundUnlaunchedRecovery = (): void => { + if (recovers === undefined || runBegan) return; + deps.fleetRecords.refundRecovery(recovers, call.id); + }; + const admission = deps.admission ?? getProcessAdmissionQueue(); const childCtl = new AbortController(); deps.sessions.registerCancel(session.id, () => { + refundUnlaunchedRecovery(); admission.cancel(session.id); if (!childCtl.signal.aborted) childCtl.abort(); }); @@ -1374,6 +1495,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { const start = (): void => { if (!stillAdmissible()) { + refundUnlaunchedRecovery(); admission.release(session.id); return; } @@ -1383,6 +1505,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { if (!stillAdmissible()) { // Marked in flight but will never start: settle through the normal // terminal path, or the wait projection strands on "running". + refundUnlaunchedRecovery(); deps.sessions.settleRun(session.id); finalizeEnd(); admission.release(session.id); @@ -1411,6 +1534,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { log.error("spawn_agent worktree setup failed: {error}", { error: message, }); + refundUnlaunchedRecovery(); await deps.sessions.fail(session.id, message); finalizeEnd(true); admission.release(session.id); @@ -1420,6 +1544,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { if (!stillAdmissible()) { // The interrupt landed while worktree setup was in flight. + refundUnlaunchedRecovery(); deps.sessions.settleRun(session.id); finalizeEnd(); await reclaimWorktree(); @@ -1628,6 +1753,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { }; // Fire and forget: this handler must return before the worker finishes. + runBegan = true; deps .run(params) .then((result) => { @@ -1711,7 +1837,14 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { start, }); if (status === "queued") deps.fleetRecords.markQueued(session.id); - return fleetResult(call.id, fleetJson({ agent_id: session.id, status })); + return fleetResult( + call.id, + fleetJson({ + agent_id: session.id, + status, + ...(recovers !== undefined ? { recovers, attempt: 2 } : {}), + }), + ); }, }); } diff --git a/src/subagent/fleet-dry-drive.ts b/src/subagent/fleet-dry-drive.ts index 263e41e2b..5d1bcc30f 100644 --- a/src/subagent/fleet-dry-drive.ts +++ b/src/subagent/fleet-dry-drive.ts @@ -144,13 +144,12 @@ export interface MailboxWorkerDigest { } /** - * Single-successor guidance for a failed+continuable entry. The marker is - * advisory only — no runtime auto-retry backs it. + * Guidance for a failed+continuable entry. spawn_agent enforces the "at most + * once": a second recovers call gets the first replacement back. */ -export const RECOVERABLE_FAILURE_CONTINUE_GUIDANCE = - "This worker failed with a transient provider error (retryable/timeout/overload) " + - "and is terminal — do not re-wait it. You may spawn at most one successor with " + - "the same brief; do not retry in a loop."; +function recoverableFailureContinueGuidance(agentId: string): string { + return `Recoverable: call spawn_agent with recovers="${agentId}" at most once.`; +} export function shouldDriveOpenTasks(input: { previousRunning?: number | undefined; @@ -441,7 +440,7 @@ export function projectMailboxRecord( ...(taken.status === "failed" && taken.recoverableFailure === true ? { continuable: true as const, - continue_with: RECOVERABLE_FAILURE_CONTINUE_GUIDANCE, + continue_with: recoverableFailureContinueGuidance(id), } : {}), ...(taken.stopReason !== undefined diff --git a/src/subagent/fleet-recovery.test.ts b/src/subagent/fleet-recovery.test.ts new file mode 100644 index 000000000..d5b581e01 --- /dev/null +++ b/src/subagent/fleet-recovery.test.ts @@ -0,0 +1,631 @@ +import { execFileSync } from "node:child_process"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, test } from "bun:test"; +import type { ReactorEmittedEvent } from "@intx/inference"; + +import { createSpawnAgentTool } from "./agent-fleet.js"; +import { createAdmissionQueue } from "./admission.js"; +import { createResolvedProviderFailureError } from "../inference-error-message.js"; +import { + createDeniedCallEnvelope, + getProcessWorkerGrantStore, +} from "../permission/worker-grant.js"; +import { driveOpenTasksAfterFleetDry } from "./fleet-dry-drive.js"; +import { driveMailboxMail } from "./mailbox-mail-drive.js"; +import { + callFleetTool, + callFleetToolRaw, + createFleetDeps, + fleetTools, + spawnAgentId, + waitUntilMailboxTerminal, +} from "./fleet-test-harness.js"; +import { createSubAgentSessionStore } from "./session-store.js"; +import { createSubAgentLoopGuardError } from "./terminal-failure.js"; +import type { RunSubAgentParams, RunSubAgentResult } from "./types.js"; + +type Behavior = (params: RunSubAgentParams) => Promise; +type Row = Record & { + agent_id: string; + status: string; + failure?: { + failure_class?: string; + attempt?: number; + recovers?: string; + handoff?: string; + recovery?: Record; + }; +}; + +const brief = { description: "plan", prompt: "write the plan", intent: "plan" }; + +function retryable(): Error { + return createResolvedProviderFailureError("test-provider", { + category: "timeout", + message: "upstream timed out", + }); +} + +function toolStart(name: string, id: string): ReactorEmittedEvent { + return { + type: "tool.start", + seq: 1, + data: { call: { name, id, arguments: {} } }, + } as unknown as ReactorEmittedEvent; +} + +const failRetryably: Behavior = async () => { + throw retryable(); +}; + +function failAfterTools(names: readonly string[]): Behavior { + return async (params) => { + names.forEach((name, i) => params.onEvent?.(toolStart(name, `t${i}`))); + throw retryable(); + }; +} + +const succeed: Behavior = async () => ({ report: "plan written" }); + +/** Fleet whose Nth spawn runs behaviors[N]; extra spawns fail the test. */ +function scriptedFleet(behaviors: readonly Behavior[]) { + const runs: RunSubAgentParams[] = []; + const deps = createFleetDeps(async (params) => { + runs.push(params); + const behavior = behaviors[runs.length - 1]; + if (behavior === undefined) throw new Error("unexpected extra spawn"); + return behavior(params); + }); + return { deps, runs, tools: fleetTools(deps) }; +} + +async function failedLane( + fleet: ReturnType, +): Promise<{ id: string; row: Row }> { + const id = await spawnAgentId(fleet.tools.spawn, brief); + await waitUntilMailboxTerminal( + fleet.deps.fleetRecords, + fleet.deps.sessions, + id, + ); + const waited = await callFleetTool(fleet.tools.wait, { + targets: [id], + timeout_ms: 5_000, + }); + const row = (waited.results as Row[])[0]; + if (row === undefined) throw new Error("no wait row"); + return { id, row }; +} + +async function recover( + fleet: ReturnType, + failedId: string, +): Promise<{ content: string; isError?: boolean }> { + return callFleetToolRaw(fleet.tools.spawn, { ...brief, recovers: failedId }); +} + +function refusalReason(content: string): string | undefined { + return /^Error: recovery_unavailable reason=([a-z_]+):/.exec(content)?.[1]; +} + +describe("recovery is never automatic", () => { + test("a retryable failure yields zero replacements without a recovers call, even after fleet ticks and dry-drive", async () => { + const fleet = scriptedFleet([failRetryably]); + const { id, row } = await failedLane(fleet); + expect(row.continuable).toBe(true); + expect(row.continue_with).toBe( + `Recoverable: call spawn_agent with recovers="${id}" at most once.`, + ); + + await driveMailboxMail({ + parentProcessing: false, + mailbox: fleet.deps.fleetRecords, + lanes: fleet.deps.sessions.list(), + beginSystemContinuation: () => undefined, + send: () => ({ status: "accepted" }), + }); + await driveOpenTasksAfterFleetDry({ + previousRunning: 1, + running: 0, + openTasks: [{ id: "t1", title: "plan", status: "doing" }], + parentProcessing: false, + mailbox: fleet.deps.fleetRecords, + lanes: fleet.deps.sessions.list(), + beginSystemContinuation: () => undefined, + send: () => ({ status: "accepted" }), + }); + await callFleetTool(fleet.tools.wait, { timeout_ms: 20 }); + await new Promise((resolve) => setTimeout(resolve, 20)); + + expect(fleet.runs).toHaveLength(1); + expect(fleet.deps.sessions.list()).toHaveLength(1); + expect(fleet.deps.sessions.get(id)?.failure?.recovery).toEqual({ + available: true, + reason: "eligible", + }); + }); + + test("a blank recovers is a fresh spawn, not a recovery request", async () => { + const fleet = scriptedFleet([succeed, succeed]); + for (const recovers of ["", " "]) { + const spawned = await callFleetToolRaw(fleet.tools.spawn, { + ...brief, + recovers, + }); + expect(spawned.isError).not.toBe(true); + expect(spawned.content).not.toContain("recovers="); + } + expect(fleet.deps.sessions.list()).toHaveLength(2); + }); +}); + +describe("enforced recovery", () => { + test("one recovers call starts exactly one replacement linked to the failure", async () => { + const fleet = scriptedFleet([failRetryably, succeed]); + const { id } = await failedLane(fleet); + + const result = await callFleetTool(fleet.tools.spawn, { + ...brief, + recovers: id, + }); + const replacementId = result.agent_id as string; + expect(result).toMatchObject({ recovers: id, attempt: 2 }); + expect(replacementId).not.toBe(id); + await waitUntilMailboxTerminal( + fleet.deps.fleetRecords, + fleet.deps.sessions, + replacementId, + ); + + expect(fleet.runs).toHaveLength(2); + expect(fleet.deps.sessions.get(replacementId)?.recovers).toBe(id); + expect(fleet.deps.fleetRecords.peek(replacementId)?.status).toBe("done"); + + // The failed lane stays failed; it only gained the replacement link. + const failed = fleet.deps.sessions.get(id); + expect(failed?.lifecycle.state).toBe("failed"); + expect(fleet.deps.fleetRecords.peek(id)?.status).toBe("failed"); + expect(failed?.failure).toMatchObject({ + failure_class: "provider_retryable", + attempt: 1, + recovery: { + available: false, + reason: "already_recovered", + replacement_id: replacementId, + }, + }); + }); + + test("concurrent recovers calls converge on one replacement", async () => { + const fleet = scriptedFleet([failRetryably, succeed]); + const { id } = await failedLane(fleet); + + const [a, b] = await Promise.all([recover(fleet, id), recover(fleet, id)]); + const first = JSON.parse(a.content) as Record; + const second = JSON.parse(b.content) as Record; + expect(a.isError).not.toBe(true); + expect(b.isError).not.toBe(true); + expect(second.agent_id).toBe(first.agent_id); + expect(second).toMatchObject({ + recovers: id, + status: "recovery_already_started", + reason: "already_recovered", + }); + + const third = await callFleetTool(fleet.tools.spawn, { + ...brief, + recovers: id, + }); + expect(third.agent_id).toBe(first.agent_id); + expect(third.reason).toBe("already_recovered"); + + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(fleet.runs).toHaveLength(2); + expect(fleet.deps.sessions.list()).toHaveLength(2); + }); + + test("a failed replacement is recovery_exhausted and cannot be recovered again", async () => { + const fleet = scriptedFleet([failRetryably, failRetryably]); + const { id } = await failedLane(fleet); + const replacementId = ( + await callFleetTool(fleet.tools.spawn, { ...brief, recovers: id }) + ).agent_id as string; + await waitUntilMailboxTerminal( + fleet.deps.fleetRecords, + fleet.deps.sessions, + replacementId, + ); + const waited = await callFleetTool(fleet.tools.wait, { + targets: [replacementId], + timeout_ms: 5_000, + }); + const row = (waited.results as Row[])[0]; + expect(row?.status).toBe("failed"); + expect(row?.failure).toMatchObject({ + failure_class: "provider_retryable", + attempt: 2, + recovers: id, + recovery: { available: false, reason: "recovery_exhausted" }, + }); + expect(row?.continuable).toBeUndefined(); + expect(row?.continue_with).toBeUndefined(); + + const again = await recover(fleet, replacementId); + expect(again.isError).toBe(true); + expect(refusalReason(again.content)).toBe("recovery_exhausted"); + expect(fleet.runs).toHaveLength(2); + }); + + test("a failure whose session was evicted is still recovered exactly once", async () => { + const runs: RunSubAgentParams[] = []; + const behaviors = [failRetryably, succeed]; + const deps = createFleetDeps( + async (params) => { + runs.push(params); + const behavior = behaviors[runs.length - 1]; + if (behavior === undefined) throw new Error("unexpected extra spawn"); + return behavior(params); + }, + { sessions: createSubAgentSessionStore({ maxCompleted: 0 }) }, + ); + const fleet = { deps, runs, tools: fleetTools(deps) }; + const { id } = await failedLane(fleet); + expect(deps.sessions.get(id)).toBeUndefined(); + expect(deps.fleetRecords.peek(id)?.failure?.recovery.available).toBe(true); + + const first = await callFleetTool(fleet.tools.spawn, { + ...brief, + recovers: id, + }); + const again = await callFleetTool(fleet.tools.spawn, { + ...brief, + recovers: id, + }); + expect(again.agent_id).toBe(first.agent_id); + expect(deps.fleetRecords.peek(id)?.failure?.recovery).toEqual({ + available: false, + reason: "already_recovered", + replacement_id: first.agent_id as string, + }); + expect(runs).toHaveLength(2); + }); + + test("an invalid spawn does not consume the recovery", async () => { + const fleet = scriptedFleet([failRetryably, succeed]); + const { id } = await failedLane(fleet); + const invalid = await callFleetToolRaw(fleet.tools.spawn, { + description: "implement", + prompt: "do it", + intent: "implement", + recovers: id, + }); + expect(invalid.isError).toBe(true); + expect(fleet.deps.sessions.get(id)?.failure?.recovery.available).toBe(true); + + const ok = await callFleetTool(fleet.tools.spawn, { + ...brief, + recovers: id, + }); + expect(ok.attempt).toBe(2); + expect(fleet.runs).toHaveLength(2); + }); +}); + +describe("a recovery grant survives a replacement that never launches", () => { + const scratch: string[] = []; + afterEach(() => { + for (const dir of scratch.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + } + }); + + function scratchDir(): string { + const dir = mkdtempSync(join(tmpdir(), "fleet-recovery-")); + scratch.push(dir); + return dir; + } + + function gitRepo(): string { + const dir = scratchDir(); + const git = (...args: string[]) => + execFileSync("git", args, { cwd: dir, stdio: "ignore" }); + git("init", "-q"); + git( + "-c", + "user.name=t", + "-c", + "user.email=t@t", + "commit", + "-q", + "--allow-empty", + "-m", + "init", + ); + return dir; + } + + async function expectRecoverable( + fleet: ReturnType, + failedId: string, + ): Promise { + const recovery = fleet.deps.sessions.get(failedId)?.failure?.recovery; + expect(recovery).toEqual({ available: true, reason: "eligible" }); + expect(fleet.deps.fleetRecords.peek(failedId)?.failure?.recovery).toEqual({ + available: true, + reason: "eligible", + }); + + const retry = await callFleetTool(fleet.tools.spawn, { + ...brief, + recovers: failedId, + }); + expect(retry).toMatchObject({ recovers: failedId, attempt: 2 }); + expect(retry.status).not.toBe("recovery_already_started"); + } + + test("a worktree setup failure leaves the lane recoverable", async () => { + const fleet = scriptedFleet([failRetryably, succeed]); + const { id } = await failedLane(fleet); + + // Not a git repo, so worktree creation fails before the run starts. + const spawnInWorktree = createSpawnAgentTool({ + ...fleet.deps, + cwd: scratchDir(), + useWorktree: true, + }); + const replacementId = ( + await callFleetTool(spawnInWorktree, { ...brief, recovers: id }) + ).agent_id as string; + await waitUntilMailboxTerminal( + fleet.deps.fleetRecords, + fleet.deps.sessions, + replacementId, + ); + expect(fleet.deps.fleetRecords.peek(replacementId)?.status).toBe("failed"); + expect(fleet.runs).toHaveLength(1); + + await expectRecoverable(fleet, id); + }); + + test("a cancel that lands during worktree setup leaves the lane recoverable", async () => { + const fleet = scriptedFleet([failRetryably, succeed]); + const { id } = await failedLane(fleet); + + const spawnInWorktree = createSpawnAgentTool({ + ...fleet.deps, + cwd: gitRepo(), + useWorktree: true, + }); + const replacementId = ( + await callFleetTool(spawnInWorktree, { ...brief, recovers: id }) + ).agent_id as string; + fleet.deps.sessions.cancel(replacementId); + await new Promise((resolve) => setTimeout(resolve, 200)); + expect(fleet.runs).toHaveLength(1); + + await expectRecoverable(fleet, id); + }); + + test("cancelling a replacement still queued for admission leaves the lane recoverable", async () => { + const fleet = scriptedFleet([failRetryably, succeed]); + const { id } = await failedLane(fleet); + + const spawnQueued = createSpawnAgentTool({ + ...fleet.deps, + admission: createAdmissionQueue({ capacity: 0 }), + }); + const replacementId = ( + await callFleetTool(spawnQueued, { ...brief, recovers: id }) + ).agent_id as string; + fleet.deps.sessions.cancel(replacementId); + expect(fleet.runs).toHaveLength(1); + + await expectRecoverable(fleet, id); + }); + + test("a replacement that ran and failed still spends the grant", async () => { + const fleet = scriptedFleet([failRetryably, failRetryably]); + const { id } = await failedLane(fleet); + const replacementId = ( + await callFleetTool(fleet.tools.spawn, { ...brief, recovers: id }) + ).agent_id as string; + await waitUntilMailboxTerminal( + fleet.deps.fleetRecords, + fleet.deps.sessions, + replacementId, + ); + expect(fleet.deps.sessions.get(id)?.failure?.recovery).toMatchObject({ + available: false, + reason: "already_recovered", + replacement_id: replacementId, + }); + }); +}); + +describe("ineligible recovery spawns nothing", () => { + test.each([ + ["a completed write", ["read_file", "write_file"]], + ["a run_shell call", ["grep", "run_shell"]], + ["an MCP tool", ["mcp__tracker__create_issue"]], + ["an unknown tool", ["mystery_tool"]], + ] as const)("side_effects_completed after %s", async (_label, names) => { + const fleet = scriptedFleet([failAfterTools(names)]); + const { id, row } = await failedLane(fleet); + expect(row.failure?.failure_class).toBe("provider_retryable"); + expect(row.failure?.recovery).toEqual({ + available: false, + reason: "side_effects_completed", + }); + expect(row.continuable).toBeUndefined(); + expect(row.continue_with).toBeUndefined(); + + const refused = await recover(fleet, id); + expect(refused.isError).toBe(true); + expect(refusalReason(refused.content)).toBe("side_effects_completed"); + expect(fleet.runs).toHaveLength(1); + expect(fleet.deps.sessions.list()).toHaveLength(1); + }); + + test("read, search, and web read calls keep a failure replay-safe", async () => { + const fleet = scriptedFleet([ + failAfterTools(["read_file", "grep", "search_files", "web_fetch"]), + ]); + const { row } = await failedLane(fleet); + expect(row.failure?.recovery).toEqual({ + available: true, + reason: "eligible", + }); + expect(row.continuable).toBe(true); + }); + + test.each([ + [ + "provider_fatal", + async (): Promise => { + throw createResolvedProviderFailureError("test-provider", { + category: "credential_failure", + message: "Authentication failed", + }); + }, + ], + [ + "loop_guard", + async (): Promise => { + throw createSubAgentLoopGuardError( + new Error("reactor error: Doom loop detected: x"), + ); + }, + ], + ] as const)("not_retryable for %s", async (failureClass, behavior) => { + const fleet = scriptedFleet([behavior]); + const { id, row } = await failedLane(fleet); + expect(row.failure?.failure_class).toBe(failureClass); + const refused = await recover(fleet, id); + expect(refused.isError).toBe(true); + expect(refusalReason(refused.content)).toBe("not_retryable"); + expect(fleet.runs).toHaveLength(1); + }); + + test("not_retryable for a cancelled worker", async () => { + const fleet = scriptedFleet([() => new Promise(() => undefined)]); + const id = await spawnAgentId(fleet.tools.spawn, brief); + fleet.deps.sessions.cancel(id); + fleet.deps.sessions.settleRun(id); + expect(fleet.deps.sessions.get(id)?.failure?.failure_class).toBe( + "cancelled", + ); + const refused = await recover(fleet, id); + expect(refused.isError).toBe(true); + expect(refusalReason(refused.content)).toBe("not_retryable"); + expect(fleet.runs).toHaveLength(1); + }); + + test("not_retryable for a worker that is still running", async () => { + const fleet = scriptedFleet([() => new Promise(() => undefined)]); + const id = await spawnAgentId(fleet.tools.spawn, brief); + const refused = await recover(fleet, id); + expect(refusalReason(refused.content)).toBe("not_retryable"); + expect(fleet.runs).toHaveLength(1); + }); + + test("an id that is not one of the caller's workers is refused", async () => { + const fleet = scriptedFleet([]); + const refused = await recover(fleet, "someone-else"); + expect(refused.isError).toBe(true); + expect(refused.content).toContain("not one of your workers"); + expect(fleet.runs).toHaveLength(0); + expect(fleet.deps.sessions.list()).toHaveLength(0); + }); +}); + +describe("replacement authority", () => { + const touched: string[] = []; + afterEach(() => { + for (const id of touched.splice(0)) { + getProcessWorkerGrantStore().invalidateSession(id, "test cleanup"); + } + }); + + // Callbacks and the per-run identity differ by construction; everything + // else in the params is the authority a worker runs with. + function authorityOf(params: RunSubAgentParams | undefined) { + if (params === undefined) throw new Error("run never started"); + const { + id: _id, + signal: _signal, + onEvent: _onEvent, + onRunSettled: _onRunSettled, + onAgentReady: _onAgentReady, + askDirectorPort: _askDirectorPort, + onProgress: _onProgress, + ...authority + } = params; + return authority; + } + + test("a replacement runs with exactly a fresh spawn's authority and none of the failed worker's state", async () => { + let failedAsk: Promise | undefined; + const failWithGrantAndAsk: Behavior = async (params) => { + const workerId = params.id ?? ""; + touched.push(workerId); + getProcessWorkerGrantStore().register( + createDeniedCallEnvelope({ + callId: "denied-1", + tool: "write_file", + subject: "src/a.ts", + args: { path: "src/a.ts" }, + cwd: params.cwd, + workerSessionId: workerId, + }), + ); + params.onAgentReady?.({ + close: async () => undefined, + interrupt: () => undefined, + followup: async () => "", + deliver: () => undefined, + }); + failedAsk = params.askDirectorPort?.register({ + question: "may I write src/a.ts?", + questionId: "ask-1", + assessment: { + policyVersion: "1", + classification: "director_resolvable", + blockedOutcome: "test blocked outcome", + unavailableDirectorPath: "test director path", + permittedAlternatives: [], + minimumAddition: "test minimum addition", + declineConsequence: "test decline consequence", + }, + }); + void failedAsk?.catch(() => undefined); + throw retryable(); + }; + const fleet = scriptedFleet([failWithGrantAndAsk, succeed, succeed]); + const { id, row } = await failedLane(fleet); + expect(row.failure?.handoff).toBe("unavailable"); + await expect(failedAsk).rejects.toThrow("session failed"); + + const replacementId = ( + await callFleetTool(fleet.tools.spawn, { ...brief, recovers: id }) + ).agent_id as string; + const freshId = await spawnAgentId(fleet.tools.spawn, brief); + touched.push(replacementId, freshId); + await waitUntilMailboxTerminal( + fleet.deps.fleetRecords, + fleet.deps.sessions, + freshId, + ); + + const replacementRun = fleet.runs.find((p) => p.id === replacementId); + const freshRun = fleet.runs.find((p) => p.id === freshId); + expect(authorityOf(replacementRun)).toEqual(authorityOf(freshRun)); + expect(replacementRun?.permissionGate).toBe(freshRun?.permissionGate); + + const grants = getProcessWorkerGrantStore(); + expect(grants.pendingForSession(id)).toBeUndefined(); + expect(grants.pendingForSession(replacementId)).toBeUndefined(); + expect(fleet.deps.sessions.hasPendingAsk(replacementId)).toBe(false); + }); +}); diff --git a/src/subagent/session-store-failure-record.test.ts b/src/subagent/session-store-failure-record.test.ts index a84293dac..ba59d4f98 100644 --- a/src/subagent/session-store-failure-record.test.ts +++ b/src/subagent/session-store-failure-record.test.ts @@ -554,3 +554,38 @@ describe("backstop close honesty", () => { expect(sessions.get(id)?.failure?.cleanup).toBe("partial"); }); }); + +describe("recovery replacement link", () => { + function failedWorker() { + const { sessions } = store(); + const id = startRunning(sessions); + sessions.fail(id, "upstream overloaded", { + failure_class: "provider_retryable", + }); + return { sessions, id }; + } + + test("the first link wins and a second claim cannot overwrite it", () => { + const { sessions, id } = failedWorker(); + sessions.linkRecoveryReplacement(id, "r1"); + sessions.linkRecoveryReplacement(id, "r2"); + expect(sessions.get(id)?.failure?.recovery).toEqual({ + available: false, + reason: "already_recovered", + replacement_id: "r1", + }); + }); + + test("unlinking restores availability only for the linked replacement", () => { + const { sessions, id } = failedWorker(); + sessions.linkRecoveryReplacement(id, "r1"); + sessions.unlinkRecoveryReplacement(id, "other"); + expect(sessions.get(id)?.failure?.recovery.replacement_id).toBe("r1"); + + sessions.unlinkRecoveryReplacement(id, "r1"); + expect(sessions.get(id)?.failure?.recovery).toEqual({ + available: true, + reason: "eligible", + }); + }); +}); diff --git a/src/subagent/session-store.ts b/src/subagent/session-store.ts index 5062f7ba7..ad9fec28a 100644 --- a/src/subagent/session-store.ts +++ b/src/subagent/session-store.ts @@ -36,6 +36,8 @@ import { isReplaySafeToolName, isSubAgentLoopGuardError, recoveryFor, + withoutRecoveryReplacement, + withRecoveryReplacement, type SubAgentFailureClass, type SubAgentFailureInput, type SubAgentFailureRecord, @@ -160,6 +162,8 @@ export interface SubAgentSession { * mutation that makes the outcome observable; absent while the lane is live. */ failure?: SubAgentFailureRecord; + /** Set on a recovery replacement: the failed agent id it recovers. */ + recovers?: string; } export interface StartSessionInput { @@ -177,6 +181,8 @@ export interface StartSessionInput { provider?: string; /** Canonical tool names this worker hard-required at dispatch. */ requiresTools?: readonly string[]; + /** The failed agent id this session was spawned to recover. */ + recovers?: string; } export interface SubAgentSessionStoreOptions { @@ -221,6 +227,17 @@ export interface SubAgentSessionStore { error: string, failure?: SubAgentFailureInput, ): Promise; + /** + * Claim a failure record for its replacement. The record otherwise stays + * as written: the failed lane never returns to running or success. + */ + linkRecoveryReplacement(id: string, replacementId: string): void; + /** + * Undo a claim whose replacement never launched. No-op unless + * `replacementId` is the one currently linked, so a stale refund cannot + * reopen a lane another replacement has since claimed. + */ + unlinkRecoveryReplacement(id: string, replacementId: string): void; // Register the live abort handle so cancel() stops the child reactor // (agent.close), not just flips status. registerCancel(id: string, abort: () => void): void; @@ -832,7 +849,7 @@ export function createSubAgentSessionStore( const id = session.id; const failureClass = terminalFailureClass(session); if (failureClass !== undefined) { - const attempt = 1; + const attempt = session.recovers !== undefined ? 2 : 1; const teardownFailed = pendingFailures.get(id)?.teardownFailed === true; session.failure = Object.freeze({ agent_id: id, @@ -851,6 +868,9 @@ export function createSubAgentSessionStore( }), ), attempt, + ...(session.recovers !== undefined + ? { recovers: session.recovers } + : {}), handoff: terminalHandoffs.get(id) ?? currentHandoff(id), cleanup: teardownFailed || releaseFaults.has(id) ? "partial" : "released", @@ -1449,6 +1469,7 @@ export function createSubAgentSessionStore( ...(input.requiresTools !== undefined ? { requiresTools: [...input.requiresTools] } : {}), + ...(input.recovers !== undefined ? { recovers: input.recovers } : {}), }; sessions.set(id, session); bumpRevision(id); @@ -1770,6 +1791,26 @@ export function createSubAgentSessionStore( return Promise.resolve(); }, + linkRecoveryReplacement(id: string, replacementId: string): void { + mutate(id, (session) => { + if (session.failure === undefined) return; + // First claim wins: a second link would orphan the replacement the + // first caller was already handed. + if (session.failure.recovery.replacement_id !== undefined) return; + session.failure = withRecoveryReplacement( + session.failure, + replacementId, + ); + }); + }, + + unlinkRecoveryReplacement(id: string, replacementId: string): void { + mutate(id, (session) => { + if (session.failure?.recovery.replacement_id !== replacementId) return; + session.failure = withoutRecoveryReplacement(session.failure); + }); + }, + registerCancel(id: string, abort: () => void): void { const session = sessions.get(id); if (session === undefined || !isLiveStrip(session.lifecycle)) return; @@ -2594,6 +2635,7 @@ function cloneSession( ? { requiresTools: [...session.requiresTools] } : {}), ...(session.failure !== undefined ? { failure: session.failure } : {}), + ...(session.recovers !== undefined ? { recovers: session.recovers } : {}), }; } diff --git a/src/subagent/terminal-failure.ts b/src/subagent/terminal-failure.ts index 0e4986393..47e156805 100644 --- a/src/subagent/terminal-failure.ts +++ b/src/subagent/terminal-failure.ts @@ -171,3 +171,35 @@ function isRetryableFailure(input: { isRecoverableProviderFailureCategory(cause.category) ); } + +/** + * The record as written, for a claim whose replacement never launched. A + * record is only claimable while recovery is available, so that is the state + * a refund restores. + */ +export function withoutRecoveryReplacement( + record: SubAgentFailureRecord, +): SubAgentFailureRecord { + return Object.freeze({ + ...record, + recovery: Object.freeze({ + available: true, + reason: "eligible" as const, + }), + }); +} + +/** The same record, claimed by a replacement. Nothing else changes. */ +export function withRecoveryReplacement( + record: SubAgentFailureRecord, + replacementId: string, +): SubAgentFailureRecord { + return Object.freeze({ + ...record, + recovery: Object.freeze({ + available: false, + reason: "already_recovered" as const, + replacement_id: replacementId, + }), + }); +}