From b7e948b9998bd6a77665062b0859096b6d6663f2 Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Sat, 10 Oct 2026 00:08:45 -0400 Subject: [PATCH 1/3] feat(subagent): record typed terminal failures for the parent A failed worker reached the parent as two booleans and error prose, so the parent could not tell a transient provider fault from a stuck or cancelled lane, and an answered ask_director left no trace once the run died. The store now writes one frozen failure record in the same mutation that makes a non-clean terminal visible, and provider_failure and continuable derive from its class so every projection agrees. --- docs/ARCHITECTURE.md | 2 +- src/subagent/agent-fleet.test.ts | 24 +- src/subagent/agent-fleet.ts | 76 +-- src/subagent/fleet-dry-drive.test.ts | 1 + src/subagent/fleet-dry-drive.ts | 6 + src/subagent/fleet-terminal-failure.test.ts | 373 ++++++++++++++ src/subagent/lifecycle-tools.test.ts | 5 + src/subagent/lifecycle-tools.ts | 8 +- .../run-resolved-provider-failure.test.ts | 2 + .../session-store-failure-record.test.ts | 483 ++++++++++++++++++ src/subagent/session-store.ts | 277 +++++++++- src/subagent/terminal-failure.test.ts | 144 ++++++ src/subagent/terminal-failure.ts | 142 +++++ src/subagent/types.ts | 2 + 14 files changed, 1472 insertions(+), 73 deletions(-) create mode 100644 src/subagent/fleet-terminal-failure.test.ts create mode 100644 src/subagent/session-store-failure-record.test.ts create mode 100644 src/subagent/terminal-failure.test.ts create mode 100644 src/subagent/terminal-failure.ts diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 61b086365..21c96d088 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`). 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`) 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. #### Model-family policy (`src/agent/model-family-policy.ts`) diff --git a/src/subagent/agent-fleet.test.ts b/src/subagent/agent-fleet.test.ts index 96f7aa5c7..27e6751e6 100644 --- a/src/subagent/agent-fleet.test.ts +++ b/src/subagent/agent-fleet.test.ts @@ -1227,9 +1227,15 @@ describe("interrupt_agent unblocks wait_agents", () => { agent_id: string; status: string; stop_reason?: string; + failure?: unknown; }[]; expect(results).toEqual([ - { agent_id: id, status: "interrupted", stop_reason: "interrupted" }, + { + agent_id: id, + status: "interrupted", + stop_reason: "interrupted", + failure: expect.objectContaining({ failure_class: "interrupted" }), + }, ]); expect(deps.sessions.get(id)?.lifecycleStatus).toBe("interrupted"); expect(deps.sessions.get(id)?.status).toBe("running"); @@ -1589,9 +1595,15 @@ describe("interrupt_agent unblocks wait_agents", () => { agent_id: string; status: string; stop_reason?: string; + failure?: unknown; }[]; expect(results).toEqual([ - { agent_id: id, status: "interrupted", stop_reason: "interrupted" }, + { + agent_id: id, + status: "interrupted", + stop_reason: "interrupted", + failure: expect.objectContaining({ failure_class: "interrupted" }), + }, ]); expect(deps.fleetRecords.peek(id)?.status).toBe("interrupted"); expect(deps.fleetRecords.peek(id)?.collected).toBe(true); @@ -2520,9 +2532,15 @@ describe("admission queue", () => { agent_id: string; status: string; stop_reason?: string; + failure?: unknown; }[]; expect(results).toEqual([ - { agent_id: queuedId, status: "interrupted", stop_reason: "cancelled" }, + { + agent_id: queuedId, + status: "interrupted", + stop_reason: "cancelled", + failure: expect.objectContaining({ failure_class: "cancelled" }), + }, ]); expect(started()).toBe(1); diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index 05c4eab66..f660d2739 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -111,9 +111,10 @@ import { import { formatSubAgentSpawnAuthFailureMessage } from "./inference-auth-failure.js"; import { - isRecoverableProviderFailureCategory, - isResolvedProviderFailureError, -} from "../inference-error-message.js"; + classifySubAgentFailure, + isProviderFailureClass, + type SubAgentFailureRecord, +} from "./terminal-failure.js"; import { errorMessage } from "../agent/error-message.js"; import { isSubAgentCancelError } from "./dispose.js"; import { @@ -132,9 +133,11 @@ interface FleetRecord { report?: string; error?: string; stopReason?: string; + /** Derived from `failure.failure_class`. */ providerFailure?: true; - /** Transient provider failure — the parent may spawn one successor. */ + /** Transient provider failure, derived from `failure.recovery.available`. */ recoverableFailure?: true; + failure?: SubAgentFailureRecord; /** Set once a wait_agents caller has been handed this result. */ collected?: boolean; /** Set once a waiter or occupancy take handed report/error. */ @@ -169,9 +172,8 @@ interface FleetOverlay { bodyHanded?: boolean; tombstoned?: boolean; hint?: string; - providerFailure?: true; - /** Transient provider failure — same meaning as the FleetRecord flag. */ - recoverableFailure?: true; + /** Mirror of the session's failure record so it survives eviction. */ + failure?: SubAgentFailureRecord; } const RECOVERY_HINT = @@ -267,20 +269,6 @@ class FleetMailbox { this.sessions.wake(); } - markProviderFailure(id: string): void { - const existing = this.records.get(id); - if (existing === undefined) return; - existing.providerFailure = true; - } - - /** Stamp a transient provider failure beside sessions.fail; the snapshot - * projects it even once the payload is tombstoned. */ - markRecoverable(id: string): void { - const existing = this.records.get(id); - if (existing === undefined) return; - existing.recoverableFailure = true; - } - markQueued(id: string): void { const existing = this.records.get(id); if (existing === undefined || existing.collected === true) return; @@ -381,6 +369,7 @@ class FleetMailbox { private rememberLiveWaitStatuses(): void { for (const [id, overlay] of this.records) { + this.mirrorFailure(id, overlay); const wait = this.sessionWaitStatus(id); if (wait === undefined) continue; // Keep lastWait while admission-queued; the live projection would @@ -390,6 +379,13 @@ class FleetMailbox { } } + private mirrorFailure(id: string, overlay: FleetOverlay): void { + const session = this.sessions.get(id); + if (session === undefined) return; + if (session.failure !== undefined) overlay.failure = session.failure; + else delete overlay.failure; + } + private waitStatusFromEvicted(id: string): WaitJSONStatus | undefined { return waitStatusFromVerbLifecycle(this.sessions.evictedLifecycle(id)); } @@ -430,6 +426,7 @@ class FleetMailbox { return { status: "running" }; } const session = this.sessions.get(id); + this.mirrorFailure(id, overlay); if ( session === undefined && overlay.tombstoned !== true && @@ -457,6 +454,7 @@ class FleetMailbox { : payload?.stopReason; const ask = status === "awaiting_director" ? this.sessions.peekAsk(id) : undefined; + const failure = isLiveWaitStatus(status) ? undefined : overlay.failure; return { status, ...(overlay.collected === true ? { collected: true } : {}), @@ -468,10 +466,13 @@ class FleetMailbox { ? { error: payload.error } : {}), ...(stopReason !== undefined ? { stopReason } : {}), - ...(overlay.providerFailure === true ? { providerFailure: true } : {}), - ...(overlay.recoverableFailure === true + ...(failure !== undefined && isProviderFailureClass(failure.failure_class) + ? { providerFailure: true } + : {}), + ...(failure?.recovery.available === true ? { recoverableFailure: true } : {}), + ...(failure !== undefined ? { failure } : {}), ...(ask !== undefined ? { question: ask.question, @@ -1409,7 +1410,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { log.error("spawn_agent worktree setup failed: {error}", { error: message, }); - deps.sessions.fail(session.id, message); + await deps.sessions.fail(session.id, message); finalizeEnd(true); admission.release(session.id); return; @@ -1660,7 +1661,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { : {}), }); }) - .catch((err) => { + .catch(async (err) => { const alreadyCancelled = deps.sessions.get(session.id)?.status === "cancelled"; if ( @@ -1673,25 +1674,23 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { deps.sessions.settleRun(session.id); return; } - const isProviderFailure = isResolvedProviderFailureError(err); const diagnosticMessage = errorMessage(err); const authMessage = formatSubAgentSpawnAuthFailureMessage( description, err, ); const failReason = authMessage ?? diagnosticMessage; - if (isProviderFailure || providerFailureObserved) { - deps.fleetRecords.markProviderFailure(session.id); - } - // A transient failure stays wait-terminal failed but carries a - // continuable marker so the parent can spawn one successor. - if ( - isResolvedProviderFailureError(err) && - isRecoverableProviderFailureCategory(err.category) - ) { - deps.fleetRecords.markRecoverable(session.id); - } - deps.sessions.fail(session.id, failReason); + // CL-10202: the typed class is the single source for the + // provider_failure and continuable projections. A transient + // failure stays wait-terminal failed; nothing here retries. + await deps.sessions.fail(session.id, failReason, { + failure_class: classifySubAgentFailure(err, { + providerFailureObserved, + }), + ...(settlement?.teardown_failed === true + ? { teardownFailed: true } + : {}), + }); }) .finally(() => { activeLanes.delete(call.id); @@ -2031,6 +2030,7 @@ export function createListAgentsTool(deps: WaitAgentsDeps): AgentTool { } : {}), ...(stopReason !== undefined ? { stop_reason: stopReason } : {}), + ...(record?.failure !== undefined ? { failure: record.failure } : {}), ...(record?.status === "awaiting_director" && record.question !== undefined ? { question: record.question } diff --git a/src/subagent/fleet-dry-drive.test.ts b/src/subagent/fleet-dry-drive.test.ts index 66b413b17..0619a58d2 100644 --- a/src/subagent/fleet-dry-drive.test.ts +++ b/src/subagent/fleet-dry-drive.test.ts @@ -263,6 +263,7 @@ describe("collectUncollectedTerminals", () => { status: "failed", description: "failed lane", error: "boom", + failure: expect.objectContaining({ failure_class: "error" }), }); expect(mailbox.peek("done")?.collected).toBe(true); expect(mailbox.peek("fail")?.collected).toBe(true); diff --git a/src/subagent/fleet-dry-drive.ts b/src/subagent/fleet-dry-drive.ts index cccc2288a..263e41e2b 100644 --- a/src/subagent/fleet-dry-drive.ts +++ b/src/subagent/fleet-dry-drive.ts @@ -11,6 +11,7 @@ import { } from "../plugins/result-truncation-plugin.js"; import { isLiveWaitStatus, type WaitJSONStatus } from "./lifecycle.js"; import { parseSubAgentReport } from "./report.js"; +import type { SubAgentFailureRecord } from "./terminal-failure.js"; /** Enough of a lane report for a parent continuation; traces stay on disk. */ export const FLEET_DRY_REPORT_CHARS = 8_192; @@ -44,6 +45,7 @@ export interface FleetDryMailboxRecord { /** Transient provider failure — the parent may spawn one successor. */ readonly recoverableFailure?: true; readonly stopReason?: string; + readonly failure?: SubAgentFailureRecord; } export interface FleetDryMailbox { @@ -118,6 +120,7 @@ export interface CollectedWorkerReport { continuable?: true; continue_with?: string; stop_reason?: string; + failure?: SubAgentFailureRecord; } /** Mailbox parent payload: envelope digest plus a blob pointer, not the full report. */ @@ -137,6 +140,7 @@ export interface MailboxWorkerDigest { continuable?: true; continue_with?: string; stop_reason?: string; + failure?: SubAgentFailureRecord; } /** @@ -344,6 +348,7 @@ export async function digestCollectedReport( ...(report.stop_reason !== undefined ? { stop_reason: report.stop_reason } : {}), + ...(report.failure !== undefined ? { failure: report.failure } : {}), }; } @@ -442,6 +447,7 @@ export function projectMailboxRecord( ...(taken.stopReason !== undefined ? { stop_reason: taken.stopReason } : {}), + ...(taken.failure !== undefined ? { failure: taken.failure } : {}), }; } diff --git a/src/subagent/fleet-terminal-failure.test.ts b/src/subagent/fleet-terminal-failure.test.ts new file mode 100644 index 000000000..f0e4736e0 --- /dev/null +++ b/src/subagent/fleet-terminal-failure.test.ts @@ -0,0 +1,373 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { ReactorEmittedEvent } from "@intx/inference"; + +import { createResolvedProviderFailureError } from "../inference-error-message.js"; +import { createAdmissionQueue } from "./admission.js"; +import { + callFleetTool, + createFleetDeps, + deferred, + fleetTools, + spawnAgentId, + waitUntilMailboxTerminal, +} from "./fleet-test-harness.js"; +import { pollUntil } from "./run-test-harness.js"; +import type { RunSubAgentParams, RunSubAgentResult } from "./types.js"; + +type Row = Record & { + agent_id: string; + status: string; + failure?: Record; +}; + +function retryable(): Error { + return createResolvedProviderFailureError("test-provider", { + category: "retryable", + message: "upstream overloaded", + statusCode: 429, + }); +} + +function credential(): Error { + return createResolvedProviderFailureError("test-provider", { + category: "credential_failure", + message: "Authentication failed", + statusCode: 401, + }); +} + +const inferenceError = { + type: "inference.error", + seq: 1, + data: { error: { category: "retryable" }, partial: { text: "" } }, +} as unknown as ReactorEmittedEvent; + +async function waitRows( + tools: ReturnType, + targets: string[], +): Promise { + const waited = await callFleetTool(tools.wait, { + targets, + timeout_ms: 5_000, + mode: "all", + }); + expect(waited.timed_out).toBe(false); + return waited.results as Row[]; +} + +const spawnArgs = { description: "job", prompt: "do it", intent: "explore" }; + +describe("failure_class drives the existing provider markers", () => { + test.each([ + ["retryable", retryable, "provider_retryable", true], + ["credential", credential, "provider_fatal", false], + ] as const)( + "%s failure: continuable equals recovery.available", + async (_label, makeError, failureClass, available) => { + const deps = createFleetDeps(async () => { + throw makeError(); + }); + const tools = fleetTools(deps); + const id = await spawnAgentId(tools.spawn, spawnArgs); + const [row] = await waitRows(tools, [id]); + expect(row?.status).toBe("failed"); + expect(row?.provider_failure).toBe(true); + expect(row?.failure?.failure_class).toBe(failureClass); + expect(row?.failure?.recovery).toEqual( + available + ? { available: true, reason: "eligible" } + : { available: false, reason: "not_retryable" }, + ); + expect(row?.continuable === true).toBe(available); + }, + ); + + test("a permission suspension after an observed inference.error is error, not provider", async () => { + const deps = createFleetDeps(async (params) => { + params.onEvent?.(inferenceError); + throw Object.assign( + new Error("Sub-agent send returned a suspended result"), + { + suspendedType: "suspended", + correlationId: "corr-1", + }, + ); + }); + const tools = fleetTools(deps); + const id = await spawnAgentId(tools.spawn, spawnArgs); + const [row] = await waitRows(tools, [id]); + expect(row?.failure?.failure_class).toBe("error"); + expect(row?.provider_failure).toBeUndefined(); + expect(row?.continuable).toBeUndefined(); + }); + + test("an unclassified throw after an observed inference.error stays provider_fatal", async () => { + const deps = createFleetDeps(async (params) => { + params.onEvent?.(inferenceError); + throw new Error("send rejected"); + }); + const tools = fleetTools(deps); + const id = await spawnAgentId(tools.spawn, spawnArgs); + const [row] = await waitRows(tools, [id]); + expect(row?.provider_failure).toBe(true); + expect(row?.failure?.failure_class).toBe("provider_fatal"); + expect(row?.continuable).toBeUndefined(); + }); +}); + +describe("parent-side setup failure", () => { + const dirs: string[] = []; + afterEach(async () => { + for (const dir of dirs.splice(0)) { + await rm(dir, { recursive: true, force: true }); + } + }); + + test("a worktree setup failure classifies as error, never provider", async () => { + const notARepo = await mkdtemp(join(tmpdir(), "corbits-not-a-repo-")); + const workdirBase = await mkdtemp(join(tmpdir(), "corbits-workdir-")); + dirs.push(notARepo, workdirBase); + let ran = false; + const deps = createFleetDeps( + async () => { + ran = true; + return { report: "no" }; + }, + { cwd: notARepo }, + ); + deps.getWorkdirBase = () => workdirBase; + deps.useWorktree = true; + const tools = fleetTools(deps); + const id = await spawnAgentId(tools.spawn, spawnArgs); + const [row] = await waitRows(tools, [id]); + expect(ran).toBe(false); + expect(row?.status).toBe("failed"); + expect(row?.failure?.failure_class).toBe("error"); + expect(row?.provider_failure).toBeUndefined(); + }); +}); + +describe("terminal failure cleanup", () => { + test("a worker whose teardown timed out frees its admission slot without a second close wait", async () => { + const admission = createAdmissionQueue({ capacity: 1 }); + let started = 0; + const deps = createFleetDeps( + async (params: RunSubAgentParams): Promise => { + started += 1; + if (started > 1) return { report: "second done" }; + params.onAgentReady?.({ + close: () => new Promise(() => undefined), + interrupt: () => undefined, + followup: async () => "", + deliver: () => undefined, + }); + params.onRunSettled?.({ + turn_count: 1, + input_tokens: 0, + output_tokens: 0, + cache_read_tokens: 0, + cache_write_tokens: 0, + reasoning_tokens: 0, + tool_call_count: 0, + tool_error_count: 0, + error_count: 1, + duration_ms: 1, + model: "test-model", + terminal_reason: "error", + teardown_failed: true, + }); + throw retryable(); + }, + ); + deps.admission = admission; + const tools = fleetTools(deps); + const first = await spawnAgentId(tools.spawn, spawnArgs); + const second = await callFleetTool(tools.spawn, spawnArgs); + expect(second.status).toBe("queued"); + + await waitUntilMailboxTerminal(deps.fleetRecords, deps.sessions, first); + const [row] = await waitRows(tools, [first]); + expect(row?.status).toBe("failed"); + expect(row?.failure).toMatchObject({ cleanup: "partial" }); + await pollUntil(() => !admission.occupied(first)); + + const secondId = second.agent_id as string; + await waitUntilMailboxTerminal(deps.fleetRecords, deps.sessions, secondId); + expect(deps.fleetRecords.peek(secondId)?.status).toBe("done"); + }); + + test("a failure with an open ask releases the wait, the ask, and the admission slot once", async () => { + const admission = createAdmissionQueue({ capacity: 1 }); + const failFirst = deferred(); + let started = 0; + let askOutcome: Promise | undefined; + const deps = createFleetDeps( + async (params: RunSubAgentParams): Promise => { + started += 1; + if (started > 1) return { report: "second done" }; + params.onAgentReady?.({ + close: async () => undefined, + interrupt: () => undefined, + followup: async () => "", + deliver: () => undefined, + }); + askOutcome = params.askDirectorPort?.register({ + question: "which path?", + 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 askOutcome?.catch(() => undefined); + await failFirst.promise; + throw retryable(); + }, + ); + deps.admission = admission; + const tools = fleetTools(deps); + const first = await spawnAgentId(tools.spawn, spawnArgs); + const second = await callFleetTool(tools.spawn, spawnArgs); + expect(second.status).toBe("queued"); + await pollUntil(() => deps.sessions.hasPendingAsk(first)); + + failFirst.resolve(undefined); + await waitUntilMailboxTerminal(deps.fleetRecords, deps.sessions, first); + const [row] = await waitRows(tools, [first]); + expect(row?.status).toBe("failed"); + expect(row?.failure).toMatchObject({ + failure_class: "provider_retryable", + handoff: "unavailable", + cleanup: "released", + }); + await expect(askOutcome).rejects.toThrow("session failed"); + expect(deps.sessions.hasPendingAsk(first)).toBe(false); + expect(deps.sessions.isRunInFlight(first)).toBe(false); + await pollUntil(() => !admission.occupied(first)); + + const secondId = second.agent_id as string; + await waitUntilMailboxTerminal(deps.fleetRecords, deps.sessions, secondId); + expect(started).toBe(2); + expect(deps.fleetRecords.peek(secondId)?.status).toBe("done"); + + // Collected once: a later wait does not hand the failure out again. + const again = await callFleetTool(tools.wait, { timeout_ms: 20 }); + expect((again.results as Row[]).map((r) => r.agent_id)).toEqual([secondId]); + }); + + test("late events after settlement change neither the record nor spawn anything", async () => { + let started = 0; + const deps = createFleetDeps(async () => { + started += 1; + throw retryable(); + }); + const tools = fleetTools(deps); + const id = await spawnAgentId(tools.spawn, spawnArgs); + const [row] = await waitRows(tools, [id]); + const record = deps.sessions.get(id)?.failure; + expect(record).toBeDefined(); + + deps.sessions.appendEvent(id, inferenceError); + deps.sessions.fail(id, "late", { failure_class: "loop_guard" }); + deps.sessions.complete(id, "late report"); + deps.sessions.cancel(id); + + expect(deps.sessions.get(id)?.failure).toBe( + record as NonNullable, + ); + expect(deps.fleetRecords.peek(id)?.failure).toEqual( + row?.failure as unknown as typeof record, + ); + expect(started).toBe(1); + }); +}); + +describe("parent-facing status", () => { + test("list_agents distinguishes failed, cancelled, completed, and running lanes", async () => { + const hold = deferred(); + const runs = [ + async (): Promise => { + throw retryable(); + }, + async (): Promise => ({ report: "fine" }), + (): Promise => hold.promise, + (): Promise => new Promise(() => undefined), + ]; + let next = 0; + const deps = createFleetDeps(() => { + const run = runs[next]; + next += 1; + if (run === undefined) throw new Error("unexpected spawn"); + return run(); + }); + const tools = fleetTools(deps); + const failed = await spawnAgentId(tools.spawn, spawnArgs); + const done = await spawnAgentId(tools.spawn, spawnArgs); + const cancelled = await spawnAgentId(tools.spawn, spawnArgs); + const running = await spawnAgentId(tools.spawn, spawnArgs); + await waitUntilMailboxTerminal(deps.fleetRecords, deps.sessions, failed); + await waitUntilMailboxTerminal(deps.fleetRecords, deps.sessions, done); + deps.sessions.cancel(cancelled); + deps.sessions.settleRun(cancelled); + + const listed = await callFleetTool(tools.list, {}); + const rows = new Map( + (listed.agents as Row[]).map((row) => [row.agent_id, row]), + ); + expect(rows.get(failed)?.status).toBe("failed"); + expect(rows.get(failed)?.failure?.failure_class).toBe("provider_retryable"); + expect(rows.get(cancelled)?.status).toBe("interrupted"); + expect(rows.get(cancelled)?.stop_reason).toBe("cancelled"); + expect(rows.get(cancelled)?.failure?.failure_class).toBe("cancelled"); + expect(rows.get(done)?.status).toBe("done"); + expect(rows.get(done)?.failure).toBeUndefined(); + expect(rows.get(running)?.status).toBe("running"); + expect(rows.get(running)?.failure).toBeUndefined(); + }); + + test("a stalled salvage is done with stop_reason and a stalled record", async () => { + const deps = createFleetDeps(async () => ({ + report: "salvaged", + stopReason: "stalled", + })); + const tools = fleetTools(deps); + const id = await spawnAgentId(tools.spawn, spawnArgs); + const [row] = await waitRows(tools, [id]); + expect(row?.status).toBe("done"); + expect(row?.stop_reason).toBe("stalled"); + expect(row?.failure?.failure_class).toBe("stalled"); + expect(row?.continuable).toBeUndefined(); + }); + + test("the mailbox digest carries the same record as wait_agents", async () => { + const deps = createFleetDeps(async () => { + throw retryable(); + }); + const tools = fleetTools(deps); + const id = await spawnAgentId(tools.spawn, spawnArgs); + await waitUntilMailboxTerminal(deps.fleetRecords, deps.sessions, id); + const { driveMailboxMail } = await import("./mailbox-mail-drive.js"); + const prompts: string[] = []; + await driveMailboxMail({ + parentProcessing: false, + mailbox: deps.fleetRecords, + lanes: deps.sessions.list(), + beginSystemContinuation: () => undefined, + send: (prompt: string) => { + prompts.push(prompt); + return { status: "accepted" }; + }, + }); + expect(prompts).toHaveLength(1); + expect(prompts[0]).toContain('"failure_class":"provider_retryable"'); + expect(prompts[0]).toContain('"continuable":true'); + }); +}); diff --git a/src/subagent/lifecycle-tools.test.ts b/src/subagent/lifecycle-tools.test.ts index 21c4c2a54..6ca0bfc62 100644 --- a/src/subagent/lifecycle-tools.test.ts +++ b/src/subagent/lifecycle-tools.test.ts @@ -541,11 +541,16 @@ describe("resume_agent", () => { agent_id: string; status: string; error?: string; + failure?: unknown; }[]; expect(results[0]).toEqual({ agent_id: worker.id, status: "failed", error: "resumed turn failed", + failure: expect.objectContaining({ + failure_class: "error", + recovery: { available: false, reason: "not_retryable" }, + }), }); }); diff --git a/src/subagent/lifecycle-tools.ts b/src/subagent/lifecycle-tools.ts index 3b6ea99af..9cacd8a9f 100644 --- a/src/subagent/lifecycle-tools.ts +++ b/src/subagent/lifecycle-tools.ts @@ -21,6 +21,7 @@ import { UNSTRUCTURED_DIGEST_CHARS, } from "./fleet-dry-drive.js"; import { parseSubAgentReport } from "./report.js"; +import { classifySubAgentFailure } from "./terminal-failure.js"; import { DEFAULT_MAX_ENTRY_CHARS, type AgentLifecycleStatus, @@ -349,9 +350,10 @@ export function createResumeAgentTool(deps: ResumeAgentToolDeps): AgentTool { onReply: (reply) => { deps.sessions.complete(target, reply); }, - onFail: (err) => { - deps.sessions.fail(target, errorMessage(err)); - }, + onFail: (err) => + deps.sessions.fail(target, errorMessage(err), { + failure_class: classifySubAgentFailure(err), + }), }); if (!outcome.ok) { const hint = outcome.hint !== undefined ? ` ${outcome.hint}` : ""; diff --git a/src/subagent/run-resolved-provider-failure.test.ts b/src/subagent/run-resolved-provider-failure.test.ts index 0985c49b9..035c82673 100644 --- a/src/subagent/run-resolved-provider-failure.test.ts +++ b/src/subagent/run-resolved-provider-failure.test.ts @@ -323,6 +323,7 @@ describe("resolved sub-agent provider failures", () => { status?: string; error?: string; provider_failure?: boolean; + failure?: unknown; }[]; }; @@ -331,6 +332,7 @@ describe("resolved sub-agent provider failures", () => { status: "failed", error: SAFE_MESSAGE, provider_failure: true, + failure: expect.objectContaining({ failure_class: "provider_fatal" }), }); const serialized = JSON.stringify(waitResult); expect(serialized).not.toContain(RAW_DIAGNOSTIC); diff --git a/src/subagent/session-store-failure-record.test.ts b/src/subagent/session-store-failure-record.test.ts new file mode 100644 index 000000000..18ac811b6 --- /dev/null +++ b/src/subagent/session-store-failure-record.test.ts @@ -0,0 +1,483 @@ +import { describe, expect, test } from "bun:test"; +import type { ReactorEmittedEvent } from "@intx/inference"; +import { AgentClosedError } from "@intx/agent"; + +import { createResolvedProviderFailureError } from "../inference-error-message.js"; +import { createSubAgentSessionStore } from "./session-store.js"; +import { classifySubAgentFailure } from "./terminal-failure.js"; + +const settleTick = () => new Promise((resolve) => setTimeout(resolve, 0)); + +const T0 = Date.parse("2026-10-09T12:00:00.000Z"); + +function store() { + let clock = T0; + const sessions = createSubAgentSessionStore({ + now: () => clock, + }); + return { + sessions, + advance: (ms: number) => { + clock += ms; + }, + }; +} + +function startRunning( + sessions: ReturnType, + id = "w1", + opts: { retained?: boolean } = {}, +): string { + sessions.start({ + id, + description: "worker", + agentId: "explorer", + brief: "b", + ...(opts.retained === true ? { retained: true } : {}), + }); + sessions.markRunInFlight(id); + sessions.markRunning(id); + return id; +} + +function ask( + sessions: ReturnType, + id: string, + opts: { reject?: (reason: unknown) => void } = {}, +): Promise { + let resolveAnswer: (answer: string) => void = () => undefined; + let rejectAnswer: (reason: unknown) => void = () => undefined; + const answer = new Promise((resolve, reject) => { + resolveAnswer = resolve; + rejectAnswer = reject; + }); + expect( + sessions.registerAsk(id, { + question: "which path?", + 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", + }, + resolve: resolveAnswer, + reject: opts.reject ?? rejectAnswer, + }), + ).toBe(true); + return answer; +} + +function toolStart(name: string, id: string): ReactorEmittedEvent { + return { + type: "tool.start", + seq: 1, + data: { call: { name, id, arguments: {} } }, + } as unknown as ReactorEmittedEvent; +} + +function toolDone( + callId: string, + content: string, + isError = false, +): ReactorEmittedEvent { + return { + type: "tool.done", + seq: 2, + data: { result: { callId, content, isError } }, + } as unknown as ReactorEmittedEvent; +} + +describe("terminal failure record", () => { + test("fail writes one frozen record before subscribers can observe the failure", () => { + const { sessions, advance } = store(); + const id = startRunning(sessions); + advance(1_500); + const seenWithoutRecord: string[] = []; + sessions.subscribe(() => { + const snap = sessions.get(id); + if (snap?.lifecycle.state === "failed" && snap.failure === undefined) { + seenWithoutRecord.push(id); + } + }); + + sessions.fail(id, "upstream overloaded", { + failure_class: "provider_retryable", + }); + + expect(seenWithoutRecord).toEqual([]); + const snap = sessions.get(id); + expect(snap?.failure).toEqual({ + agent_id: id, + failure_class: "provider_retryable", + failed_at: new Date(T0 + 1_500).toISOString(), + recovery: { available: true, reason: "eligible" }, + attempt: 1, + handoff: "none", + cleanup: "released", + }); + expect(snap?.failure?.failed_at).toBe( + new Date(snap?.finishedAt ?? 0).toISOString(), + ); + expect(Object.isFrozen(snap?.failure)).toBe(true); + }); + + test("fail without a classification records error", () => { + const { sessions } = store(); + const id = startRunning(sessions); + sessions.fail(id, "boom"); + expect(sessions.get(id)?.failure?.failure_class).toBe("error"); + expect(sessions.get(id)?.failure?.recovery).toEqual({ + available: false, + reason: "not_retryable", + }); + }); + + test("late fail, complete, and tool events after settlement change nothing", () => { + const { sessions, advance } = store(); + const id = startRunning(sessions); + sessions.fail(id, "first", { failure_class: "provider_retryable" }); + const record = sessions.get(id)?.failure; + expect(record).toBeDefined(); + + advance(10); + sessions.fail(id, "second", { failure_class: "loop_guard" }); + sessions.complete(id, "late report"); + sessions.appendEvent(id, toolStart("ask_director", "c1")); + sessions.appendEvent(id, toolDone("c1", "late answer")); + sessions.cancel(id); + + const after = sessions.get(id); + expect(after?.failure).toBe(record as NonNullable); + expect(after?.lifecycle.state).toBe("failed"); + expect(after?.error).toBe("first"); + }); + + test("a teardown failure reports partial cleanup", () => { + const { sessions } = store(); + const id = startRunning(sessions); + sessions.fail(id, "session close exceeded 10ms", { + failure_class: "error", + teardownFailed: true, + }); + expect(sessions.get(id)?.failure?.cleanup).toBe("partial"); + }); + + test("an ask release that throws reports partial cleanup", () => { + const { sessions } = store(); + const id = startRunning(sessions); + void ask(sessions, id, { + reject: () => { + throw new Error("reject hook exploded"); + }, + }); + sessions.fail(id, "boom", { failure_class: "error" }); + expect(sessions.get(id)?.failure?.cleanup).toBe("partial"); + }); + + test("a failure on a follow-up turn is never recovery-eligible", async () => { + const { sessions } = store(); + const id = startRunning(sessions, "w1", { retained: true }); + sessions.registerFollowup(id, () => new Promise(() => undefined)); + sessions.complete(id, "first report"); + expect(sessions.resumeOne(id, "next").ok).toBe(true); + sessions.fail(id, "timeout", { failure_class: "provider_retryable" }); + expect(sessions.get(id)?.failure?.failure_class).toBe("provider_retryable"); + expect(sessions.get(id)?.failure?.recovery).toEqual({ + available: false, + reason: "not_retryable", + }); + }); + + test("a follow-up rejected because the agent closed records error, not interrupted", async () => { + const { sessions } = store(); + const id = startRunning(sessions, "w1", { retained: true }); + sessions.registerFollowup(id, async () => { + throw new AgentClosedError(); + }); + sessions.complete(id, "first report"); + expect(sessions.resumeOne(id, "next").ok).toBe(true); + await new Promise((resolve) => setTimeout(resolve, 0)); + const snap = sessions.get(id); + expect(snap?.lifecycle.state).toBe("failed"); + expect(snap?.failure?.failure_class).toBe("error"); + expect(snap?.failure?.recovery.reason).toBe("not_retryable"); + }); +}); + +describe("terminal record for forced stops", () => { + test.each([ + ["stalled", "stalled"], + ["deadline", "deadline"], + ["incomplete-report", "incomplete_report"], + ["cancelled", "cancelled"], + ] as const)("a %s salvage completion records %s", (stopReason, expected) => { + const { sessions } = store(); + const id = startRunning(sessions); + sessions.complete(id, "salvage", { agentRetained: false, stopReason }); + expect(sessions.get(id)?.failure?.failure_class).toBe(expected); + expect(sessions.get(id)?.failure?.recovery.available).toBe(false); + }); + + test("a clean completion has no record", () => { + const { sessions } = store(); + const id = startRunning(sessions); + sessions.complete(id, "done"); + expect(sessions.get(id)?.failure).toBeUndefined(); + }); + + test("operator cancel records cancelled", () => { + const { sessions } = store(); + const id = startRunning(sessions); + expect(sessions.cancel(id)).toBe(true); + expect(sessions.get(id)?.failure?.failure_class).toBe("cancelled"); + }); + + test("an interrupted lane records interrupted and drops it when resumed", () => { + const { sessions } = store(); + const id = startRunning(sessions, "w1", { retained: true }); + sessions.registerInterrupt(id, () => undefined); + sessions.registerFollowup(id, () => new Promise(() => undefined)); + expect(sessions.interruptOne(id).ok).toBe(true); + sessions.attachReport(id, "salvage", { stopReason: "interrupted" }); + expect(sessions.get(id)?.failure?.failure_class).toBe("interrupted"); + + expect(sessions.resumeOne(id, "carry on").ok).toBe(true); + expect(sessions.get(id)?.lifecycle.state).toBe("running"); + expect(sessions.get(id)?.failure).toBeUndefined(); + }); +}); + +describe("handoff state", () => { + test("no ask in the final turn gives none", () => { + const { sessions } = store(); + const id = startRunning(sessions); + sessions.fail(id, "boom", { failure_class: "provider_retryable" }); + expect(sessions.get(id)?.failure?.handoff).toBe("none"); + }); + + test("an ask still open at termination gives unavailable and is released", async () => { + const { sessions } = store(); + const id = startRunning(sessions); + const answer = ask(sessions, id); + sessions.fail(id, "boom", { failure_class: "provider_retryable" }); + expect(sessions.get(id)?.failure?.handoff).toBe("unavailable"); + expect(sessions.hasPendingAsk(id)).toBe(false); + expect(sessions.isRunInFlight(id)).toBe(false); + await expect(answer).rejects.toThrow("session failed"); + }); + + test("an answered ask the run never consumed gives submitted", () => { + const { sessions } = store(); + const id = startRunning(sessions); + sessions.appendEvent(id, toolStart("ask_director", "c1")); + void ask(sessions, id); + expect(sessions.sendInputOne(id, "use src/").ok).toBe(true); + sessions.fail(id, "timeout", { failure_class: "provider_retryable" }); + expect(sessions.get(id)?.failure?.handoff).toBe("submitted"); + }); + + test("resolveAsk without consumption also gives submitted", () => { + const { sessions } = store(); + const id = startRunning(sessions); + void ask(sessions, id); + expect(sessions.resolveAsk(id, "use src/")).toBe(true); + sessions.fail(id, "timeout", { failure_class: "provider_retryable" }); + expect(sessions.get(id)?.failure?.handoff).toBe("submitted"); + }); + + test("delivered only once the answered ask_director call returns into the run", () => { + const { sessions } = store(); + const id = startRunning(sessions); + sessions.appendEvent(id, toolStart("ask_director", "c1")); + sessions.appendEvent(id, toolStart("read_file", "c2")); + void ask(sessions, id); + expect(sessions.sendInputOne(id, "use src/").ok).toBe(true); + // A different tool finishing is not evidence the answer was consumed. + sessions.appendEvent(id, toolDone("c2", "file body")); + sessions.appendEvent(id, toolDone("c1", "use src/")); + sessions.fail(id, "timeout", { failure_class: "provider_retryable" }); + expect(sessions.get(id)?.failure?.handoff).toBe("delivered"); + }); + + test("an errored ask_director result does not count as delivery", () => { + const { sessions } = store(); + const id = startRunning(sessions); + sessions.appendEvent(id, toolStart("ask_director", "c1")); + void ask(sessions, id); + expect(sessions.sendInputOne(id, "use src/").ok).toBe(true); + sessions.appendEvent(id, toolDone("c1", "Error: lost", true)); + sessions.fail(id, "timeout", { failure_class: "provider_retryable" }); + expect(sessions.get(id)?.failure?.handoff).toBe("submitted"); + }); + + test("an ask_director return without a submitted answer never claims delivery", () => { + const { sessions } = store(); + const id = startRunning(sessions); + sessions.appendEvent(id, toolStart("ask_director", "c1")); + sessions.appendEvent( + id, + toolDone("c1", "Error: ask_director was cancelled."), + ); + sessions.fail(id, "boom", { failure_class: "error" }); + expect(sessions.get(id)?.failure?.handoff).toBe("none"); + }); + + test("a follow-up turn starts with a fresh handoff trace", () => { + const { sessions } = store(); + const id = startRunning(sessions, "w1", { retained: true }); + sessions.registerFollowup(id, () => new Promise(() => undefined)); + void ask(sessions, id); + expect(sessions.sendInputOne(id, "answer").ok).toBe(true); + sessions.complete(id, "first report"); + expect(sessions.resumeOne(id, "next").ok).toBe(true); + sessions.fail(id, "boom", { failure_class: "error" }); + expect(sessions.get(id)?.failure?.handoff).toBe("none"); + }); +}); + +describe("follow-up failure classification", () => { + const providerError = () => + createResolvedProviderFailureError("test-provider", { + category: "retryable", + message: "provider said retryable", + }); + + test("a provider failure classifies the same via send_input interrupt and resume_agent", async () => { + const viaSendInput = store().sessions; + const sendId = startRunning(viaSendInput, "w1", { retained: true }); + viaSendInput.registerInterrupt(sendId, () => undefined); + viaSendInput.registerFollowup(sendId, async () => { + throw providerError(); + }); + expect( + viaSendInput.sendInputOne(sendId, "steer", { interrupt: true }).ok, + ).toBe(true); + viaSendInput.attachReport(sendId, "salvage", { stopReason: "interrupted" }); + await settleTick(); + + const viaResume = store().sessions; + const resumeId = startRunning(viaResume, "w1", { retained: true }); + viaResume.registerFollowup(resumeId, async () => { + throw providerError(); + }); + viaResume.complete(resumeId, "first report"); + expect( + viaResume.resumeOne(resumeId, "next", { + onFail: (err) => { + viaResume.fail(resumeId, "provider said retryable", { + failure_class: classifySubAgentFailure(err), + }); + }, + }).ok, + ).toBe(true); + await settleTick(); + + expect(viaResume.get(resumeId)?.failure?.failure_class).toBe( + "provider_retryable", + ); + expect(viaSendInput.get(sendId)?.failure?.failure_class).toBe( + viaResume.get(resumeId)?.failure?.failure_class, + ); + }); + + test("a non-provider follow-up failure via send_input interrupt records error", async () => { + const { sessions } = store(); + const id = startRunning(sessions, "w1", { retained: true }); + sessions.registerInterrupt(id, () => undefined); + sessions.registerFollowup(id, async () => { + throw new Error("boom"); + }); + sessions.sendInputOne(id, "steer", { interrupt: true }); + sessions.attachReport(id, "salvage", { stopReason: "interrupted" }); + await settleTick(); + expect(sessions.get(id)?.failure?.failure_class).toBe("error"); + }); +}); + +describe("backstop close honesty", () => { + function retainedWithClose( + sessions: ReturnType, + close: (deadlineMs?: number) => Promise, + ): string { + const id = startRunning(sessions, "w1", { retained: true }); + sessions.registerClose(id, close); + return id; + } + + test("a clean backstop close records released cleanup", async () => { + const { sessions } = store(); + const id = retainedWithClose(sessions, async () => undefined); + await sessions.fail(id, "boom", { failure_class: "error" }); + expect(sessions.get(id)?.failure?.cleanup).toBe("released"); + }); + + test("a rejecting backstop close records partial cleanup", async () => { + const { sessions } = store(); + const id = retainedWithClose(sessions, async () => { + throw new Error("child still live"); + }); + await sessions.fail(id, "boom", { failure_class: "error" }); + expect(sessions.get(id)?.lifecycle.state).toBe("failed"); + expect(sessions.get(id)?.failure?.cleanup).toBe("partial"); + }); + + test("a backstop close that hangs past its deadline records partial cleanup", async () => { + const sessions = createSubAgentSessionStore({ + now: () => T0, + backstopCloseDeadlineMs: 20, + }); + const id = retainedWithClose( + sessions, + () => new Promise(() => undefined), + ); + await sessions.fail(id, "boom", { failure_class: "error" }); + expect(sessions.get(id)?.failure?.cleanup).toBe("partial"); + }); + + test("the record is not written until the close settles, and never changes after", async () => { + const { sessions } = store(); + let finishClose: () => void = () => undefined; + const id = retainedWithClose( + sessions, + () => + new Promise((resolve) => { + finishClose = resolve; + }), + ); + const failing = sessions.fail(id, "boom", { failure_class: "error" }); + expect(sessions.get(id)?.failure).toBeUndefined(); + finishClose(); + await failing; + const record = sessions.get(id)?.failure; + expect(record?.cleanup).toBe("released"); + await settleTick(); + expect(sessions.get(id)?.failure).toBe( + record as NonNullable, + ); + }); + + test("a worker whose own teardown already failed settles without a second close wait", async () => { + // Default (production) backstop deadline: waiting on it would stall the + // lane, and the admission slot behind it, for another 30s. + const { sessions } = store(); + const id = retainedWithClose( + sessions, + () => new Promise(() => undefined), + ); + const settled = await Promise.race([ + sessions + .fail(id, "boom", { failure_class: "error", teardownFailed: true }) + .then(() => "settled" as const), + new Promise<"stalled">((resolve) => + setTimeout(() => resolve("stalled"), 200), + ), + ]); + expect(settled).toBe("settled"); + expect(sessions.get(id)?.lifecycle.state).toBe("failed"); + expect(sessions.get(id)?.failure?.cleanup).toBe("partial"); + }); +}); diff --git a/src/subagent/session-store.ts b/src/subagent/session-store.ts index 368eddd9a..9ef47e51f 100644 --- a/src/subagent/session-store.ts +++ b/src/subagent/session-store.ts @@ -29,6 +29,16 @@ import type { VerifiedOutcome, } from "./escalation-policy.js"; import { deriveTerminalOutcome } from "./escalation-policy.js"; +import { canonicalToolName } from "../agent/canonical-tool-name.js"; +import { + classifySubAgentFailure, + failureClassForStopReason, + recoveryFor, + type SubAgentFailureClass, + type SubAgentFailureInput, + type SubAgentFailureRecord, + type SubAgentHandoffState, +} from "./terminal-failure.js"; const log = getLogger([LOG_NAMESPACE_ROOT, "subagent", "session-store"]); @@ -46,6 +56,9 @@ async function invokeCloseBounded( } } +/** resume_agent's awaits the backstop close, so a handler may be async. */ +type FollowupFailHandler = (error: unknown) => void | Promise; + export type SubAgentSessionStatus = "running" | "done" | "failed" | "cancelled"; /** Lifecycle for close_agent/resume_agent; snapshot projects @@ -140,6 +153,11 @@ export interface SubAgentSession { /** Canonical tool names hard-required at dispatch. Verified pre-spawn; the * mount re-checks and fails a stale snapshot if one went missing. */ requiresTools?: readonly string[]; + /** + * Typed record of a terminal non-clean outcome. Written once, in the same + * mutation that makes the outcome observable; absent while the lane is live. + */ + failure?: SubAgentFailureRecord; } export interface StartSessionInput { @@ -175,6 +193,10 @@ export interface SubAgentSessionStoreOptions { createId?: () => string; /** When set, resume/followup inference is admitted through this queue. */ admission?: AdmissionQueue; + // How long fail() waits on a live close before recording partial cleanup. + // Defaults to close_agent's deadline: the same teardown, so no tighter or + // looser bound is justified. + backstopCloseDeadlineMs?: number; } export interface SubAgentSessionStore { @@ -195,7 +217,13 @@ export interface SubAgentSessionStore { report: string, opts?: { agentRetained?: boolean; stopReason?: ForcedStopReason }, ): void; - fail(id: string, error: string): void; + // Resolves once the lane is terminal and its failure record is written. A + // live backstop close is awaited first so the record's cleanup is true. + fail( + id: string, + error: string, + failure?: SubAgentFailureInput, + ): Promise; // Register the live abort handle so cancel() stops the child reactor // (agent.close), not just flips status. registerCancel(id: string, abort: () => void): void; @@ -233,7 +261,7 @@ export interface SubAgentSessionStore { opts?: { onStart?: () => void; onReply?: (reply: string) => void; - onFail?: (error: unknown) => void; + onFail?: FollowupFailHandler; }, ): | { ok: true; status: "running" | "queued" } @@ -260,7 +288,7 @@ export interface SubAgentSessionStore { interrupt?: boolean; onFollowupReply?: (reply: string) => void; onStart?: () => void; - onFail?: (error: unknown) => void; + onFail?: FollowupFailHandler; }, ): | { ok: true; status: AgentLifecycleStatus } @@ -494,6 +522,8 @@ export function createSubAgentSessionStore( const now = options.now ?? (() => Date.now()); const createId = options.createId ?? defaultCreateId; const admission = options.admission; + const backstopCloseDeadlineMs = + options.backstopCloseDeadlineMs ?? DEFAULT_CLOSE_DEADLINE_MS; // Insertion order: older first. list() returns a snapshot in that order. const sessions = new Map(); @@ -527,7 +557,7 @@ export function createSubAgentSessionStore( failLifecycle: "completed" | "interrupted"; onStart?: () => void; onReply?: (reply: string) => void; - onFail?: (error: unknown) => void; + onFail?: FollowupFailHandler; } // Overlapping interrupt-steers queue FIFO per session. The head launches // when the live run settles (attachReport handoff or complete()'s @@ -553,11 +583,13 @@ export function createSubAgentSessionStore( .join(", "); for (const stashed of queue) { try { - stashed.onFail?.( + const settled = stashed.onFail?.( new Error( `send_input steer "${steerPreview(stashed.message)}" dropped (${reason})`, ), ); + // A drop has no caller to wait on an async handler. + if (settled !== undefined) void settled.catch(() => undefined); } catch { // A throwing onFail must not break session settlement. } @@ -590,6 +622,20 @@ export function createSubAgentSessionStore( deniedCall?: WorkerDeniedCallEnvelope; } >(); + // CL-10202: answering an ask deletes it, so without this trace a failure + // could not tell "answered but never consumed" from "never asked". Scoped + // to the current turn: start and every follow-up turn reset it. + const handoffTraces = new Map(); + // Handoff as it stood just before a terminal transition cancelled the + // pending ask, consumed when the failure record is written. + const terminalHandoffs = new Map(); + // Sessions whose ask cancellation or grant invalidation threw during the + // current terminal transition. + const releaseFaults = new Set(); + const pendingFailures = new Map(); + // A failure on a follow-up turn is never recovery-eligible: the spawn-time + // brief no longer describes the work the session was doing. + const followupTurns = new Set(); const listeners = new Set<() => void>(); // Tombstones for pruneRetained-evicted sessions, insertion-ordered so the // oldest drops first past MAX_EVICTED_TOMBSTONES. Lets resume_agent report @@ -686,12 +732,14 @@ export function createSubAgentSessionStore( getProcessWorkerGrantStore().invalidateSession(id, reason); } catch { // Grant invalidation must not throw into settle/interrupt paths. + releaseFaults.add(id); } } try { pending.reject(new Error(reason)); } catch { // Reject must not throw into settle/interrupt paths. + releaseFaults.add(id); } if (!silent) notify(); return true; @@ -705,11 +753,112 @@ export function createSubAgentSessionStore( } }; + const currentHandoff = (id: string): SubAgentHandoffState => + pendingAsks.has(id) ? "unavailable" : (handoffTraces.get(id) ?? "none"); + + const clearTurnTraces = (): void => { + handoffTraces.clear(); + terminalHandoffs.clear(); + releaseFaults.clear(); + pendingFailures.clear(); + followupTurns.clear(); + }; + + const resetTurnTrace = (id: string): void => { + handoffTraces.delete(id); + terminalHandoffs.delete(id); + releaseFaults.delete(id); + pendingFailures.delete(id); + }; + + // Every terminal transition cancels the pending ask here first, which + // deletes it; capture the handoff before that so the record can say an ask + // was still open. const settleCancelsAsks = (id: string, reason: string): void => { + terminalHandoffs.set(id, currentHandoff(id)); + releaseFaults.delete(id); cancelAskInternal(id, reason); cancelDescendantAsks(id, reason); }; + const FORCED_STOP_REASONS: ReadonlySet = new Set([ + "cancelled", + "deadline", + "stalled", + "incomplete-report", + "interrupted", + ]); + + const forcedStopClass = ( + session: StoredSession, + ): SubAgentFailureClass | undefined => { + const reason = session.stopReason; + return reason !== undefined && FORCED_STOP_REASONS.has(reason) + ? failureClassForStopReason(reason as ForcedStopReason) + : undefined; + }; + + const terminalFailureClass = ( + session: StoredSession, + ): SubAgentFailureClass | undefined => { + switch (session.lifecycle.state) { + case "failed": + return pendingFailures.get(session.id)?.failure_class ?? "error"; + case "cancelled": + return "cancelled"; + case "interrupted": { + // A rejected follow-up restores the lane to interrupted so + // resume_agent can retry, but the cause still decides the class. + const followupFailure = pendingFailures.get(session.id); + if (followupFailure !== undefined) return followupFailure.failure_class; + return forcedStopClass(session); + } + case "completed": + case "shutdown": + return forcedStopClass(session); + default: + return undefined; + } + }; + + // CL-10202: runs inside every mutation, before subscribers are notified, + // so no observer can see a terminal non-clean lane without its record. A + // record never coexists with a live lane: an interrupted worker that + // resumes starts a new attempt. + const reconcileTerminalRecord = (session: StoredSession): void => { + const state = session.lifecycle.state; + if (state === "pending_init" || state === "running") { + delete session.failure; + return; + } + if (session.failure !== undefined) return; + const id = session.id; + const failureClass = terminalFailureClass(session); + if (failureClass !== undefined) { + const attempt = 1; + const teardownFailed = pendingFailures.get(id)?.teardownFailed === true; + session.failure = Object.freeze({ + agent_id: id, + failure_class: failureClass, + failed_at: new Date(session.finishedAt ?? now()).toISOString(), + recovery: Object.freeze( + recoveryFor({ + failureClass, + attempt, + followupTurn: followupTurns.has(id), + }), + ), + attempt, + handoff: terminalHandoffs.get(id) ?? currentHandoff(id), + cleanup: + teardownFailed || releaseFaults.has(id) ? "partial" : "released", + }); + } + terminalHandoffs.delete(id); + releaseFaults.delete(id); + pendingFailures.delete(id); + }; + // Invariant: a non-live lifecycle never coexists with a run-in-flight // marker once no settle-capable handle remains. A soft-interrupted run // still settles through its interrupt/close/followup/deliver handles; @@ -747,6 +896,7 @@ export function createSubAgentSessionStore( content: capText(`Cancelled: ${reason}`, maxEntryChars), }); cancelHandles.delete(session.id); + reconcileTerminalRecord(session); // closeHandles are owned by releaseHandles / closeOne — dropping them // here would skip teardown for a retained mid-turn session. No // enforceSettledRunInvariant here: after cancel the marker is the live @@ -792,17 +942,22 @@ export function createSubAgentSessionStore( // descendant cannot stall the eviction caller) and cancel. Called on every // record drop, so a retained-but-idle session's agent is never simply // forgotten. + const detachClose = (id: string): Promise | undefined => { + const close = closeHandles.get(id); + if (close === undefined) return undefined; + closeHandles.delete(id); + return invokeCloseBounded(close, backstopCloseDeadlineMs); + }; + const releaseHandles = (id: string): void => { cancelAskInternal(id, "session handles released"); - const close = closeHandles.get(id); - if (close !== undefined) { - closeHandles.delete(id); - void close(DEFAULT_CLOSE_DEADLINE_MS).catch((err: unknown) => { - log.warn("session close during handle release failed: {error}", { - error: errorMessage(err), - }); + // Eviction and prune have no record left to keep honest, so the close + // stays fire-and-forget here; fail() awaits it through detachClose. + void detachClose(id)?.catch((err: unknown) => { + log.warn("session close during handle release failed: {error}", { + error: errorMessage(err), }); - } + }); cancelHandles.delete(id); interruptHandles.delete(id); followupHandles.delete(id); @@ -918,6 +1073,7 @@ export function createSubAgentSessionStore( const session = sessions.get(id); if (session === undefined) return; fn(session); + reconcileTerminalRecord(session); enforceSettledRunInvariant(id); session.lastActivityAt = now(); bumpRevision(id); @@ -930,6 +1086,8 @@ export function createSubAgentSessionStore( // strip state it started from so resume_agent can retry. interrupt_agent's // stamp wins over that restore — never rewrite interrupted to completed. const beginFollowupTurn = (id: string): void => { + resetTurnTrace(id); + followupTurns.add(id); runInFlight.add(id); mutate(id, (s) => { s.lifecycle = { state: "running" }; @@ -979,6 +1137,9 @@ export function createSubAgentSessionStore( ) { return; } + // endFollowupTurn just restored the lane, which stamped an interrupted + // record in the same synchronous transition; the failure supersedes it. + delete session.failure; session.lifecycle = { state: "failed", error }; session.finishedAt = now(); session.lastActivityAt = now(); @@ -1053,12 +1214,12 @@ export function createSubAgentSessionStore( } pruneRetained(); }; - const settleFollowupFailure = ( + const settleFollowupFailure = async ( id: string, err: unknown, failLifecycle: "completed" | "interrupted", - onFail?: (error: unknown) => void, - ): void => { + onFail?: FollowupFailHandler, + ): Promise => { // A failed steer hands off to the next queued steer; the lane stays live // across the handoff so no observer sees a gap between turns, and only // the last settlement restores it. @@ -1067,7 +1228,7 @@ export function createSubAgentSessionStore( (stashedFollowups.get(id)?.length ?? 0) > 0 ) { try { - onFail?.(err); + await onFail?.(err); } catch { // A throwing onFail must not break the handoff to the next steer. } @@ -1080,7 +1241,9 @@ export function createSubAgentSessionStore( } runInFlight.delete(id); try { - onFail?.(err); + // Awaited so a fail() still closing the worker lands before the lane is + // restored below; a restored completed lane would drop that failure. + await onFail?.(err); } catch { // A throwing onFail must not break session settlement. } @@ -1096,7 +1259,13 @@ export function createSubAgentSessionStore( log.error("followup rejected for closed agent {id}", { id }); return; } + // Only the interrupted restore stamps a record of its own; a completed + // restore stays a clean lane, as before. + if (failLifecycle === "interrupted") { + pendingFailures.set(id, { failure_class: classifySubAgentFailure(err) }); + } endFollowupTurn(id, failLifecycle); + pendingFailures.delete(id); log.error("followup turn failed for {id}: {error}", { id, error: errorMessage(err), @@ -1109,7 +1278,7 @@ export function createSubAgentSessionStore( opts?: { onStart?: () => void; onReply?: (reply: string) => void; - onFail?: (error: unknown) => void; + onFail?: FollowupFailHandler; }, ): AdmissionStatus => { const followup = followupHandles.get(id); @@ -1131,9 +1300,9 @@ export function createSubAgentSessionStore( .then((reply) => { settleFollowupReply(id, reply, opts?.onReply); }) - .catch((err: unknown) => { - settleFollowupFailure(id, err, failLifecycle, opts?.onFail); - }) + .catch((err: unknown) => + settleFollowupFailure(id, err, failLifecycle, opts?.onFail), + ) .finally(() => { if (takesSlot) queue?.release(id); }); @@ -1182,6 +1351,8 @@ export function createSubAgentSessionStore( const next = queue?.shift(); if (queue !== undefined && queue.length === 0) stashedFollowups.delete(id); if (next === undefined) return false; + resetTurnTrace(id); + followupTurns.add(id); const followup = followupHandles.get(id); // The follow-up handle can only be gone if teardown raced the handoff; // then no follow-up can inherit the run, so drop the queue loudly and @@ -1206,7 +1377,7 @@ export function createSubAgentSessionStore( settleFollowupReply(id, reply, next.onReply); }, (err: unknown) => { - settleFollowupFailure(id, err, next.failLifecycle, next.onFail); + void settleFollowupFailure(id, err, next.failLifecycle, next.onFail); }, ); }); @@ -1250,6 +1421,8 @@ export function createSubAgentSessionStore( dropStashedFollowups(id, "session replaced"); pinCounts.delete(id); runInFlight.delete(id); + resetTurnTrace(id); + followupTurns.delete(id); forgetRevision(id); const session: StoredSession = { id, @@ -1462,6 +1635,18 @@ export function createSubAgentSessionStore( maxEntryChars, ); const isError = result.isError === true; + // The answered ask_director call returning is the only proof + // the worker consumed the parent's answer. + const callName = + session.outstandingTools.find((c) => c.callId === callId)?.name ?? + name; + if ( + !isError && + handoffTraces.get(id) === "submitted" && + canonicalToolName(callName) === "ask_director" + ) { + handoffTraces.set(id, "delivered"); + } pushEntry(session, { kind: "tool_result", callId, @@ -1547,8 +1732,39 @@ export function createSubAgentSessionStore( } }, - fail(id: string, error: string): void { + async fail( + id: string, + error: string, + failure?: SubAgentFailureInput, + ): Promise { + if (failure !== undefined) pendingFailures.set(id, failure); settleCancelsAsks(id, "session failed"); + const current = sessions.get(id); + const failable = + current !== undefined && + isLiveStrip(current.lifecycle) && + current.lifecycle.state !== "cancelled"; + // The record is written once and never amended, so the close it reports + // on has to finish first. A close that rejects or outlives its deadline + // is a cleanup fault, not a released worker. + const closing = failable ? detachClose(id) : undefined; + if (closing !== undefined && failure?.teardownFailed === true) { + // The worker's own bounded teardown already timed out on this agent, + // so the record is partial either way; waiting on a second close + // would only hold the lane and its admission slot for another + // deadline. + void closing.catch((err: unknown) => { + log.warn("backstop close after failed teardown failed: {error}", { + error: errorMessage(err), + }); + }); + } else if (closing !== undefined) { + try { + await closing; + } catch { + releaseFaults.add(id); + } + } mutate(id, (session) => { if ( !isLiveStrip(session.lifecycle) || @@ -1557,8 +1773,7 @@ export function createSubAgentSessionStore( return; // Spawn-path throws already dispose in run.ts's finally; a resumed // persisted agent does not, so the live close handle is the only - // teardown. Invoke it fire-and-forget (as prune/evict) without - // marking shutdown — a disposed spawn close is idempotent. + // teardown, awaited above. A disposed spawn close is idempotent. session.lifecycle = { state: "failed", error }; session.retained = false; session.finishedAt = now(); @@ -1573,6 +1788,7 @@ export function createSubAgentSessionStore( releaseHandles(id); pruneCompleted(); }); + pendingFailures.delete(id); }, registerCancel(id: string, abort: () => void): void { @@ -1746,7 +1962,7 @@ export function createSubAgentSessionStore( interrupt?: boolean; onFollowupReply?: (reply: string) => void; onStart?: () => void; - onFail?: (error: unknown) => void; + onFail?: FollowupFailHandler; }, ): | { ok: true; status: AgentLifecycleStatus } @@ -1824,6 +2040,7 @@ export function createSubAgentSessionStore( const pending = pendingAsks.get(id); if (pending !== undefined) { pendingAsks.delete(id); + handoffTraces.set(id, "submitted"); pending.resolve(message); notify(); } @@ -1886,6 +2103,7 @@ export function createSubAgentSessionStore( const pending = pendingAsks.get(id); if (pending === undefined) return false; pendingAsks.delete(id); + handoffTraces.set(id, "submitted"); pending.resolve(answer); notify(); return true; @@ -2067,7 +2285,7 @@ export function createSubAgentSessionStore( opts?: { onStart?: () => void; onReply?: (reply: string) => void; - onFail?: (error: unknown) => void; + onFail?: FollowupFailHandler; }, ): | { ok: true; status: "running" | "queued" } @@ -2299,6 +2517,7 @@ export function createSubAgentSessionStore( revisions.clear(); snapshotCache.clear(); evicted.clear(); + clearTurnTraces(); try { getProcessWorkerGrantStore().invalidateAll("store cleared"); } catch { @@ -2341,6 +2560,7 @@ export function createSubAgentSessionStore( runInFlight.clear(); revisions.clear(); snapshotCache.clear(); + clearTurnTraces(); try { getProcessWorkerGrantStore().invalidateAll(reason); } catch { @@ -2394,6 +2614,7 @@ function cloneSession( ...(session.requiresTools !== undefined ? { requiresTools: [...session.requiresTools] } : {}), + ...(session.failure !== undefined ? { failure: session.failure } : {}), }; } diff --git a/src/subagent/terminal-failure.test.ts b/src/subagent/terminal-failure.test.ts new file mode 100644 index 000000000..4b8911b60 --- /dev/null +++ b/src/subagent/terminal-failure.test.ts @@ -0,0 +1,144 @@ +import { describe, expect, test } from "bun:test"; + +import { createResolvedProviderFailureError } from "../inference-error-message.js"; +import { + classifySubAgentFailure, + createSubAgentLoopGuardError, + failureClassForStopReason, + isProviderFailureClass, + isSubAgentLoopGuardError, + recoveryFor, +} from "./terminal-failure.js"; + +function providerError(category: string): Error { + return createResolvedProviderFailureError("test-provider", { + category, + message: `provider said ${category}`, + }); +} + +function permissionSuspension(): Error { + // Same shape assertReplySend throws when a worker send parks on a gate. + return Object.assign( + new Error("Sub-agent send returned a suspended result"), + { + suspendedType: "suspended", + correlationId: "corr-1", + }, + ); +} + +describe("classifySubAgentFailure", () => { + test.each([ + ["retryable", "provider_retryable"], + ["timeout", "provider_retryable"], + ["credential_failure", "provider_fatal"], + ["quota_exhausted", "provider_fatal"], + ["context_overflow", "provider_fatal"], + ["fatal", "provider_fatal"], + ] as const)("resolved provider category %s is %s", (category, expected) => { + expect(classifySubAgentFailure(providerError(category))).toBe(expected); + }); + + test("a loop-guard stop is loop_guard, never a provider class", () => { + const wrapped = createSubAgentLoopGuardError( + new Error("reactor error: Doom loop detected: x"), + ); + expect(classifySubAgentFailure(wrapped)).toBe("loop_guard"); + expect( + classifySubAgentFailure(wrapped, { providerFailureObserved: true }), + ).toBe("loop_guard"); + expect( + classifySubAgentFailure( + createSubAgentLoopGuardError(providerError("retryable")), + ), + ).toBe("loop_guard"); + }); + + test("the loop-guard wrapper keeps the cause's message", () => { + const cause = new Error("reactor error: Doom loop detected: x"); + const wrapped = createSubAgentLoopGuardError(cause); + expect(wrapped.message).toBe(cause.message); + expect(wrapped.cause).toBe(cause); + expect(isSubAgentLoopGuardError(wrapped)).toBe(true); + const lookalike = new Error(cause.message); + lookalike.name = "SubAgentLoopGuardError"; + expect(isSubAgentLoopGuardError(lookalike)).toBe(false); + }); + + test("a permission suspension is error even after an observed inference.error", () => { + expect( + classifySubAgentFailure(permissionSuspension(), { + providerFailureObserved: true, + }), + ).toBe("error"); + }); + + test("an unclassified throw is error unless the fleet saw inference.error last", () => { + expect(classifySubAgentFailure(new Error("boom"))).toBe("error"); + expect( + classifySubAgentFailure(new Error("boom"), { + providerFailureObserved: true, + }), + ).toBe("provider_fatal"); + }); + + test("forced-stop reasons map onto their own classes", () => { + expect(failureClassForStopReason("stalled")).toBe("stalled"); + expect(failureClassForStopReason("deadline")).toBe("deadline"); + expect(failureClassForStopReason("incomplete-report")).toBe( + "incomplete_report", + ); + expect(failureClassForStopReason("cancelled")).toBe("cancelled"); + expect(failureClassForStopReason("interrupted")).toBe("interrupted"); + }); + + test("only provider classes count as provider failures", () => { + expect(isProviderFailureClass("provider_retryable")).toBe(true); + expect(isProviderFailureClass("provider_fatal")).toBe(true); + expect(isProviderFailureClass("loop_guard")).toBe(false); + expect(isProviderFailureClass("error")).toBe(false); + }); +}); + +describe("recoveryFor", () => { + test("a retryable failure on the original spawn is eligible", () => { + expect( + recoveryFor({ + failureClass: "provider_retryable", + attempt: 1, + followupTurn: false, + }), + ).toEqual({ available: true, reason: "eligible" }); + }); + + test.each([ + "provider_fatal", + "loop_guard", + "cancelled", + "stalled", + "error", + ] as const)("%s is not_retryable", (failureClass) => { + expect( + recoveryFor({ failureClass, attempt: 1, followupTurn: false }), + ).toEqual({ available: false, reason: "not_retryable" }); + }); + + test("a retryable failure on a follow-up turn is not_retryable", () => { + expect( + recoveryFor({ + failureClass: "provider_retryable", + attempt: 1, + followupTurn: true, + }), + ).toEqual({ available: false, reason: "not_retryable" }); + }); + + 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 }), + ).toEqual({ available: false, reason: "recovery_exhausted" }); + } + }); +}); diff --git a/src/subagent/terminal-failure.ts b/src/subagent/terminal-failure.ts new file mode 100644 index 000000000..7e6658e73 --- /dev/null +++ b/src/subagent/terminal-failure.ts @@ -0,0 +1,142 @@ +import { + isRecoverableProviderFailureCategory, + isResolvedProviderFailureError, +} from "../inference-error-message.js"; +import { errorMessage } from "../agent/error-message.js"; +import type { ForcedStopReason } from "./stop-policy.js"; + +/** + * Closed classification of how a worker run ended without a clean result. + * Parents branch on this value; nothing downstream re-derives it from the + * error or report prose. + */ +export type SubAgentFailureClass = + | "provider_retryable" + | "provider_fatal" + | "loop_guard" + | "stalled" + | "deadline" + | "incomplete_report" + | "cancelled" + | "interrupted" + | "error"; + +export type SubAgentRecoveryReason = + | "eligible" + | "not_retryable" + | "side_effects_completed" + | "already_recovered" + | "recovery_exhausted"; + +/** + * Parent-input handoff at termination. `delivered` requires the worker's own + * ask_director call to have returned the answer; `unknown` is never upgraded. + */ +export type SubAgentHandoffState = + | "none" + | "unavailable" + | "submitted" + | "delivered" + | "unknown"; + +export type SubAgentCleanupOutcome = "released" | "partial"; + +export interface SubAgentRecovery { + readonly available: boolean; + readonly reason: SubAgentRecoveryReason; + readonly replacement_id?: string; +} + +/** + * Written once when a run reaches a terminal non-clean outcome, before the + * parent can observe that outcome. Snake case because it is projected to the + * parent verbatim. + */ +export interface SubAgentFailureRecord { + readonly agent_id: string; + readonly failure_class: SubAgentFailureClass; + readonly failed_at: string; + readonly recovery: SubAgentRecovery; + readonly attempt: number; + readonly recovers?: string; + readonly handoff: SubAgentHandoffState; + readonly cleanup: SubAgentCleanupOutcome; +} + +/** What the run's owner knows about a failure when it calls `fail`. */ +export interface SubAgentFailureInput { + readonly failure_class: SubAgentFailureClass; + /** The worker's own session teardown threw or timed out. */ + readonly teardownFailed?: boolean; +} + +export function isProviderFailureClass(cls: SubAgentFailureClass): boolean { + return cls === "provider_retryable" || cls === "provider_fatal"; +} + +export function failureClassForStopReason( + reason: ForcedStopReason, +): SubAgentFailureClass { + return reason === "incomplete-report" ? "incomplete_report" : reason; +} + +const loopGuardErrors = new WeakSet(); + +/** + * Typed wrapper for a run the reactor's doom-loop guard stopped. The message + * is the cause's, so operator-facing error text does not change. + */ +export function createSubAgentLoopGuardError(cause: unknown): Error { + const error = new Error(errorMessage(cause), { cause }); + error.name = "SubAgentLoopGuardError"; + loopGuardErrors.add(error); + return error; +} + +export function isSubAgentLoopGuardError(err: unknown): boolean { + return err instanceof Error && loopGuardErrors.has(err); +} + +// assertReplySend tags a send that parked on an approval gate; that is a +// permission outcome even if an inference.error happened to precede it. +function isPermissionSuspension(err: unknown): boolean { + return err instanceof Error && "suspendedType" in err; +} + +/** + * Classify a thrown worker run. Only typed signals count: the loop-guard + * wrapper, a resolved provider failure, and the fleet's own observation that + * the last inference event was an `inference.error`. + */ +export function classifySubAgentFailure( + err: unknown, + ctx: { providerFailureObserved?: boolean } = {}, +): SubAgentFailureClass { + if (isSubAgentLoopGuardError(err)) return "loop_guard"; + if (isResolvedProviderFailureError(err)) { + return isRecoverableProviderFailureCategory(err.category) + ? "provider_retryable" + : "provider_fatal"; + } + if (isPermissionSuspension(err)) return "error"; + if (ctx.providerFailureObserved === true) return "provider_fatal"; + return "error"; +} + +/** + * Recovery availability for a freshly written record. The chain cap wins + * over class: a recovery attempt that fails is exhausted whatever its cause. + */ +export function recoveryFor(input: { + failureClass: SubAgentFailureClass; + attempt: number; + followupTurn: boolean; +}): SubAgentRecovery { + if (input.attempt > 1) { + return { available: false, reason: "recovery_exhausted" }; + } + if (input.followupTurn || input.failureClass !== "provider_retryable") { + return { available: false, reason: "not_retryable" }; + } + return { available: true, reason: "eligible" }; +} diff --git a/src/subagent/types.ts b/src/subagent/types.ts index d0baee42b..31ff1ede9 100644 --- a/src/subagent/types.ts +++ b/src/subagent/types.ts @@ -285,6 +285,8 @@ export interface SubAgentRunSettlement extends SubAgentTelemetryRollup { duration_ms: number; model: string; terminal_reason: SubAgentTerminalReason; + /** The worker's own session teardown threw or exceeded its deadline. */ + teardown_failed?: true; } export interface RunSubAgentResult { From 694fbd27e0a541d6da326e0e33b983151db4476c Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Sat, 10 Oct 2026 00:08:54 -0400 Subject: [PATCH 2/3] feat(subagent): classify doom-loop worker stops as loop_guard The reactor's doom-loop guard already stops workers, but the run ended as a plain error and its class survived only in the reactor message. The doom_loop run end reaches the stream after the send rejects, so the run notes it in its stream sink and rethrows a typed error once teardown has drained the stream. A forced-stop reason would have taken the salvage path and turned the failed lane into a done one. The run also reports when its own teardown fails, which is the release step behind a partial cleanup outcome. --- src/subagent/run-loop-guard.test.ts | 150 ++++++++++++++++++++++ src/subagent/run-terminal-cleanup.test.ts | 64 +++++++++ src/subagent/run.ts | 39 +++++- 3 files changed, 249 insertions(+), 4 deletions(-) create mode 100644 src/subagent/run-loop-guard.test.ts create mode 100644 src/subagent/run-terminal-cleanup.test.ts diff --git a/src/subagent/run-loop-guard.test.ts b/src/subagent/run-loop-guard.test.ts new file mode 100644 index 000000000..74619ee94 --- /dev/null +++ b/src/subagent/run-loop-guard.test.ts @@ -0,0 +1,150 @@ +import { describe, expect, test } from "bun:test"; +import { rm } from "node:fs/promises"; +import { join } from "node:path"; +import type { ReactorEmittedEvent } from "@intx/inference"; + +import { + callFleetTool, + createFleetDeps, + deferred, + fleetTools, + spawnAgentId, +} from "./fleet-test-harness.js"; +import { + baseRunParams, + stubAgent, + tmpSubAgentCwd, + withStubbedAgent, +} from "./run-test-harness.js"; +import { isSubAgentLoopGuardError } from "./terminal-failure.js"; +import type { + RunSubAgentParams, + RunSubAgentResult, + SubAgentRunSettlement, +} from "./types.js"; + +const DOOM_MESSAGE = + "Doom loop detected: an identical tool batch (grep, glob) executed 3 times consecutively"; + +type Run = (params: RunSubAgentParams) => Promise; + +/** + * Replays the vendored reactor's doom-loop shutdown as a worker sees it: the + * fatal reactor.error rejects the send first, and the `doom_loop` run end + * only reaches the stream after that rejection. + */ +async function withRunEndingIn( + kind: string, + body: (run: Run, cwd: string) => Promise, +): Promise { + const cwd = await tmpSubAgentCwd("loop-guard-"); + const sendRejected = deferred(); + try { + return await withStubbedAgent( + stubAgent({ + send: async () => { + setTimeout(() => sendRejected.resolve(undefined), 0); + throw new Error(`reactor error: ${DOOM_MESSAGE}`); + }, + stream: () => + (async function* (): AsyncGenerator { + yield { + type: "inference.start", + seq: 1, + data: { sourceId: "test", model: "test-model", input: [] }, + } as unknown as ReactorEmittedEvent; + await sendRejected.promise; + yield { + type: "reactor.error", + seq: 2, + data: { error: DOOM_MESSAGE, fatal: true }, + } as unknown as ReactorEmittedEvent; + yield { + type: "message.run.ended", + seq: 3, + data: { + messageRunId: "run-1", + messageId: "msg-1", + status: "failed", + error: { message: DOOM_MESSAGE, kind }, + }, + } as unknown as ReactorEmittedEvent; + })(), + }), + async () => { + const { runSubAgent } = await import("./run.js"); + return body(runSubAgent, cwd); + }, + ); + } finally { + await rm(cwd, { recursive: true, force: true }); + } +} + +async function spawnAndCollect( + run: Run, + cwd: string, +): Promise> { + const deps = createFleetDeps(run, { cwd }); + deps.getWorkdirBase = () => join(cwd, ".ctx"); + const tools = fleetTools(deps); + const id = await spawnAgentId(tools.spawn, { + description: "discovery", + prompt: "find it", + intent: "explore", + }); + const waited = await callFleetTool(tools.wait, { + targets: [id], + timeout_ms: 5_000, + }); + const rows = waited.results as Record[]; + return rows[0] ?? {}; +} + +describe("loop-guard termination", () => { + test("runSubAgent rethrows a doom-loop stop as a typed loop-guard error", async () => { + await withRunEndingIn("doom_loop", async (run, cwd) => { + let settled: SubAgentRunSettlement | undefined; + let caught: unknown; + try { + await run( + baseRunParams(cwd, { + onRunSettled: (summary) => { + settled = summary; + }, + }), + ); + } catch (error) { + caught = error; + } + expect(isSubAgentLoopGuardError(caught)).toBe(true); + expect((caught as Error).message).toBe(`reactor error: ${DOOM_MESSAGE}`); + // Telemetry stop_reason is unchanged: the run still ends as "error". + expect(settled?.terminal_reason).toBe("error"); + }); + }); + + test("the parent sees loop_guard, distinct from a provider failure", async () => { + await withRunEndingIn("doom_loop", async (run, cwd) => { + const row = await spawnAndCollect(run, cwd); + expect(row.status).toBe("failed"); + expect(row.error).toBe(`reactor error: ${DOOM_MESSAGE}`); + expect(row.failure).toMatchObject({ + failure_class: "loop_guard", + recovery: { available: false, reason: "not_retryable" }, + }); + expect(row.provider_failure).toBeUndefined(); + expect(row.continuable).toBeUndefined(); + }); + }); + + test("a failed run end of another kind is not a loop guard", async () => { + await withRunEndingIn("reactor_fatal", async (run, cwd) => { + const row = await spawnAndCollect(run, cwd); + expect(row.status).toBe("failed"); + expect((row.failure as { failure_class?: string }).failure_class).toBe( + "error", + ); + }); + }); +}); diff --git a/src/subagent/run-terminal-cleanup.test.ts b/src/subagent/run-terminal-cleanup.test.ts new file mode 100644 index 000000000..f72fd68fe --- /dev/null +++ b/src/subagent/run-terminal-cleanup.test.ts @@ -0,0 +1,64 @@ +import { describe, expect, test } from "bun:test"; +import { rm } from "node:fs/promises"; +import { join } from "node:path"; + +import { + callFleetTool, + createFleetDeps, + fleetTools, + spawnAgentId, +} from "./fleet-test-harness.js"; +import { + stubAgent, + tmpSubAgentCwd, + withPosixDispose, + withStubbedAgent, +} from "./run-test-harness.js"; + +describe("terminal failure cleanup outcome", () => { + test.each([ + ["a clean teardown", async (): Promise => undefined, "released"], + [ + "a teardown that leaves children behind", + async (): Promise => { + throw new Error("1 shell child process still live after 2000ms reap"); + }, + "partial", + ], + ] as const)("%s records cleanup %s", async (_label, dispose, expected) => { + const cwd = await tmpSubAgentCwd("terminal-cleanup-"); + try { + await withPosixDispose(dispose, () => + withStubbedAgent( + stubAgent({ + send: async () => { + throw new Error("send exploded"); + }, + }), + async () => { + const { runSubAgent } = await import("./run.js"); + const deps = createFleetDeps(runSubAgent, { cwd }); + deps.getWorkdirBase = () => join(cwd, ".ctx"); + const tools = fleetTools(deps); + const id = await spawnAgentId(tools.spawn, { + description: "cleanup probe", + prompt: "fail", + intent: "explore", + }); + const waited = await callFleetTool(tools.wait, { + targets: [id], + timeout_ms: 5_000, + }); + const row = (waited.results as Record[])[0]; + expect(row?.status).toBe("failed"); + expect( + (row?.failure as { cleanup?: string } | undefined)?.cleanup, + ).toBe(expected); + }, + ), + ); + } finally { + await rm(cwd, { recursive: true, force: true }); + } + }); +}); diff --git a/src/subagent/run.ts b/src/subagent/run.ts index 22ce0ea76..bb10f40d2 100644 --- a/src/subagent/run.ts +++ b/src/subagent/run.ts @@ -207,6 +207,7 @@ import type { import type { TaskIntent } from "./report.js"; import { runWithSubAgentIdentity } from "./identity-context.js"; +import { createSubAgentLoopGuardError } from "./terminal-failure.js"; /** * Worker authorization denies without suspending, so Agent.send must settle @@ -667,7 +668,9 @@ export async function runSubAgent( }; let terminalReason: SubAgentTerminalReason = "error"; let errorCount = 0; - const settlementState = { latestModel: params.provider.model }; + const settlementState: RunSettlementState = { + latestModel: params.provider.model, + }; try { const result = await runSubAgentInner( @@ -679,8 +682,16 @@ export async function runSubAgent( return result; } catch (error) { errorCount = 1; - if (isSubAgentCancelError(error, params.signal)) + if (isSubAgentCancelError(error, params.signal)) { terminalReason = "cancelled"; + throw error; + } + // CL-10202: the inner finally has drained the stream by now, so a + // doom_loop run end the reactor emitted after rejecting the send has + // been seen. Telemetry keeps terminal_reason "error". + if (settlementState.loopGuardTripped === true) { + throw createSubAgentLoopGuardError(error); + } throw error; } finally { try { @@ -691,6 +702,9 @@ export async function runSubAgent( duration_ms: Date.now() - startedAt, model: settlementState.latestModel, terminal_reason: terminalReason, + ...(settlementState.teardownFailed === true + ? { teardown_failed: true as const } + : {}), }), ); } catch { @@ -699,10 +713,17 @@ export async function runSubAgent( } } +interface RunSettlementState { + latestModel: string; + /** The reactor ended a message run with `error.kind: "doom_loop"`. */ + loopGuardTripped?: boolean; + teardownFailed?: boolean; +} + async function runSubAgentInner( params: RunSubAgentParams, telemetryRollup: SubAgentTelemetryRollup, - settlementState: { latestModel: string }, + settlementState: RunSettlementState, ): Promise { // Fleet-spawned workers skip the pricing-cache seed re-read; the parent // runtime already applied the process seed at boot. @@ -1488,6 +1509,13 @@ async function runSubAgentInner( const runSettlement = createRunEventSettlement(); const streamSink = (rawEvent: ReactorEmittedEvent): void => { runSettlement.handleEvent(rawEvent); + if ( + rawEvent.type === "message.run.ended" && + rawEvent.data.status === "failed" && + rawEvent.data.error?.kind === "doom_loop" + ) { + settlementState.loopGuardTripped = true; + } const event = sanitizeDiagnosticValue(rawEvent, [ readSourceCredentialMaterial(workerSource.credentialId).secret, ]) as ReactorEmittedEvent; @@ -1979,7 +2007,10 @@ async function runSubAgentInner( posixTools, }), params.teardownDeadlineMs ?? DEFAULT_CLOSE_DEADLINE_MS, - ); + ).catch((teardownError: unknown) => { + settlementState.teardownFailed = true; + throw teardownError; + }); } } } From 9b1b04d00e18af072d67e9caad83cfe4127e65b1 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 12 Oct 2026 20:45:33 -0700 Subject: [PATCH 3/3] fix(subagent): keep failure records without delaying spawn failure --- src/subagent/agent-fleet.ts | 2 + src/subagent/fleet-terminal-failure.test.ts | 66 +++++++++++++++ src/subagent/lifecycle-tools.ts | 6 +- .../session-store-failure-record.test.ts | 81 +++++++++++++++---- src/subagent/session-store.ts | 46 +++-------- src/subagent/terminal-failure.test.ts | 54 +++++++++++++ src/subagent/terminal-failure.ts | 22 ++++- 7 files changed, 221 insertions(+), 56 deletions(-) diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index f660d2739..a3d07c94e 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -113,6 +113,7 @@ import { formatSubAgentSpawnAuthFailureMessage } from "./inference-auth-failure. import { classifySubAgentFailure, isProviderFailureClass, + isSubAgentLoopGuardError, type SubAgentFailureRecord, } from "./terminal-failure.js"; import { errorMessage } from "../agent/error-message.js"; @@ -1690,6 +1691,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { ...(settlement?.teardown_failed === true ? { teardownFailed: true } : {}), + ...(isSubAgentLoopGuardError(err) ? { cause: err.cause } : {}), }); }) .finally(() => { diff --git a/src/subagent/fleet-terminal-failure.test.ts b/src/subagent/fleet-terminal-failure.test.ts index f0e4736e0..00a0528c4 100644 --- a/src/subagent/fleet-terminal-failure.test.ts +++ b/src/subagent/fleet-terminal-failure.test.ts @@ -15,6 +15,7 @@ import { waitUntilMailboxTerminal, } from "./fleet-test-harness.js"; import { pollUntil } from "./run-test-harness.js"; +import { createSubAgentLoopGuardError } from "./terminal-failure.js"; import type { RunSubAgentParams, RunSubAgentResult } from "./types.js"; type Row = Record & { @@ -85,6 +86,38 @@ describe("failure_class drives the existing provider markers", () => { }, ); + test("a doom-loop over a retryable provider failure stays loop_guard and continuable", async () => { + const deps = createFleetDeps(async () => { + throw createSubAgentLoopGuardError(retryable()); + }); + const tools = fleetTools(deps); + const id = await spawnAgentId(tools.spawn, spawnArgs); + const [row] = await waitRows(tools, [id]); + expect(row?.status).toBe("failed"); + expect(row?.failure?.failure_class).toBe("loop_guard"); + expect(row?.failure?.recovery).toEqual({ + available: true, + reason: "eligible", + }); + expect(row?.provider_failure).toBeUndefined(); + expect(row?.continuable).toBe(true); + }); + + test("a doom-loop without a retryable cause stays loop_guard and not continuable", async () => { + const deps = createFleetDeps(async () => { + throw createSubAgentLoopGuardError(new Error("reactor error: doom")); + }); + const tools = fleetTools(deps); + const id = await spawnAgentId(tools.spawn, spawnArgs); + const [row] = await waitRows(tools, [id]); + expect(row?.failure?.failure_class).toBe("loop_guard"); + expect(row?.failure?.recovery).toEqual({ + available: false, + reason: "not_retryable", + }); + expect(row?.continuable).toBeUndefined(); + }); + test("a permission suspension after an observed inference.error is error, not provider", async () => { const deps = createFleetDeps(async (params) => { params.onEvent?.(inferenceError); @@ -199,6 +232,39 @@ describe("terminal failure cleanup", () => { expect(deps.fleetRecords.peek(secondId)?.status).toBe("done"); }); + test("a failed spawn with a hanging close frees its admission slot before close", async () => { + const admission = createAdmissionQueue({ capacity: 1 }); + let started = 0; + const deps = createFleetDeps( + async (params: RunSubAgentParams): Promise => { + started += 1; + if (started > 1) return { report: "second done" }; + params.onAgentReady?.({ + close: () => new Promise(() => undefined), + interrupt: () => undefined, + followup: async () => "", + deliver: () => undefined, + }); + throw new Error("spawn failed"); + }, + ); + deps.admission = admission; + const tools = fleetTools(deps); + const first = await spawnAgentId(tools.spawn, spawnArgs); + const second = await callFleetTool(tools.spawn, spawnArgs); + expect(second.status).toBe("queued"); + + await waitUntilMailboxTerminal(deps.fleetRecords, deps.sessions, first); + const [row] = await waitRows(tools, [first]); + expect(row?.status).toBe("failed"); + expect(deps.sessions.get(first)?.lifecycle.state).toBe("failed"); + await pollUntil(() => !admission.occupied(first)); + + const secondId = second.agent_id as string; + await waitUntilMailboxTerminal(deps.fleetRecords, deps.sessions, secondId); + expect(deps.fleetRecords.peek(secondId)?.status).toBe("done"); + }); + test("a failure with an open ask releases the wait, the ask, and the admission slot once", async () => { const admission = createAdmissionQueue({ capacity: 1 }); const failFirst = deferred(); diff --git a/src/subagent/lifecycle-tools.ts b/src/subagent/lifecycle-tools.ts index 9cacd8a9f..c04051af0 100644 --- a/src/subagent/lifecycle-tools.ts +++ b/src/subagent/lifecycle-tools.ts @@ -21,7 +21,10 @@ import { UNSTRUCTURED_DIGEST_CHARS, } from "./fleet-dry-drive.js"; import { parseSubAgentReport } from "./report.js"; -import { classifySubAgentFailure } from "./terminal-failure.js"; +import { + classifySubAgentFailure, + isSubAgentLoopGuardError, +} from "./terminal-failure.js"; import { DEFAULT_MAX_ENTRY_CHARS, type AgentLifecycleStatus, @@ -353,6 +356,7 @@ export function createResumeAgentTool(deps: ResumeAgentToolDeps): AgentTool { onFail: (err) => deps.sessions.fail(target, errorMessage(err), { failure_class: classifySubAgentFailure(err), + ...(isSubAgentLoopGuardError(err) ? { cause: err.cause } : {}), }), }); if (!outcome.ok) { diff --git a/src/subagent/session-store-failure-record.test.ts b/src/subagent/session-store-failure-record.test.ts index 18ac811b6..6a10edc35 100644 --- a/src/subagent/session-store-failure-record.test.ts +++ b/src/subagent/session-store-failure-record.test.ts @@ -125,6 +125,36 @@ describe("terminal failure record", () => { expect(Object.isFrozen(snap?.failure)).toBe(true); }); + test("a loop_guard with a retryable provider cause is recovery-eligible", () => { + const { sessions } = store(); + const id = startRunning(sessions); + sessions.fail(id, "upstream overloaded", { + failure_class: "loop_guard", + cause: createResolvedProviderFailureError("test-provider", { + category: "retryable", + message: "upstream overloaded", + }), + }); + const failure = sessions.get(id)?.failure; + expect(failure?.failure_class).toBe("loop_guard"); + expect(failure?.recovery).toEqual({ available: true, reason: "eligible" }); + expect("cause" in (failure ?? {})).toBe(false); + }); + + test("a loop_guard without a retryable cause is not recovery-eligible", () => { + const { sessions } = store(); + const id = startRunning(sessions); + sessions.fail(id, "doom", { + failure_class: "loop_guard", + cause: new Error("doom"), + }); + expect(sessions.get(id)?.failure?.failure_class).toBe("loop_guard"); + expect(sessions.get(id)?.failure?.recovery).toEqual({ + available: false, + reason: "not_retryable", + }); + }); + test("fail without a classification records error", () => { const { sessions } = store(); const id = startRunning(sessions); @@ -415,30 +445,47 @@ describe("backstop close honesty", () => { expect(sessions.get(id)?.failure?.cleanup).toBe("released"); }); - test("a rejecting backstop close records partial cleanup", async () => { + test("a rejecting backstop close does not delay the failed lane", async () => { const { sessions } = store(); - const id = retainedWithClose(sessions, async () => { - throw new Error("child still live"); - }); - await sessions.fail(id, "boom", { failure_class: "error" }); + let rejectClose: (err: Error) => void = () => undefined; + const id = retainedWithClose( + sessions, + () => + new Promise((_resolve, reject) => { + rejectClose = reject; + }), + ); + const failing = sessions.fail(id, "boom", { failure_class: "error" }); expect(sessions.get(id)?.lifecycle.state).toBe("failed"); - expect(sessions.get(id)?.failure?.cleanup).toBe("partial"); + const record = sessions.get(id)?.failure; + expect(record?.cleanup).toBe("released"); + rejectClose(new Error("child still live")); + await failing; + expect(sessions.get(id)?.failure).toBe( + record as NonNullable, + ); }); - test("a backstop close that hangs past its deadline records partial cleanup", async () => { - const sessions = createSubAgentSessionStore({ - now: () => T0, - backstopCloseDeadlineMs: 20, - }); + test("a hanging backstop close does not delay the failed lane", async () => { + const { sessions } = store(); const id = retainedWithClose( sessions, () => new Promise(() => undefined), ); - await sessions.fail(id, "boom", { failure_class: "error" }); - expect(sessions.get(id)?.failure?.cleanup).toBe("partial"); + const settled = await Promise.race([ + sessions + .fail(id, "boom", { failure_class: "error" }) + .then(() => "settled" as const), + new Promise<"stalled">((resolve) => + setTimeout(() => resolve("stalled"), 200), + ), + ]); + expect(settled).toBe("settled"); + expect(sessions.get(id)?.lifecycle.state).toBe("failed"); + expect(sessions.get(id)?.failure?.cleanup).toBe("released"); }); - test("the record is not written until the close settles, and never changes after", async () => { + test("the record is written before the close settles, and never changes after", async () => { const { sessions } = store(); let finishClose: () => void = () => undefined; const id = retainedWithClose( @@ -449,11 +496,11 @@ describe("backstop close honesty", () => { }), ); const failing = sessions.fail(id, "boom", { failure_class: "error" }); - expect(sessions.get(id)?.failure).toBeUndefined(); - finishClose(); - await failing; + expect(sessions.get(id)?.lifecycle.state).toBe("failed"); const record = sessions.get(id)?.failure; expect(record?.cleanup).toBe("released"); + finishClose(); + await failing; await settleTick(); expect(sessions.get(id)?.failure).toBe( record as NonNullable, diff --git a/src/subagent/session-store.ts b/src/subagent/session-store.ts index 9ef47e51f..4f386b3e8 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, + isSubAgentLoopGuardError, recoveryFor, type SubAgentFailureClass, type SubAgentFailureInput, @@ -193,9 +194,6 @@ export interface SubAgentSessionStoreOptions { createId?: () => string; /** When set, resume/followup inference is admitted through this queue. */ admission?: AdmissionQueue; - // How long fail() waits on a live close before recording partial cleanup. - // Defaults to close_agent's deadline: the same teardown, so no tighter or - // looser bound is justified. backstopCloseDeadlineMs?: number; } @@ -217,8 +215,6 @@ export interface SubAgentSessionStore { report: string, opts?: { agentRetained?: boolean; stopReason?: ForcedStopReason }, ): void; - // Resolves once the lane is terminal and its failure record is written. A - // live backstop close is awaited first so the record's cleanup is true. fail( id: string, error: string, @@ -846,6 +842,7 @@ export function createSubAgentSessionStore( failureClass, attempt, followupTurn: followupTurns.has(id), + cause: pendingFailures.get(id)?.cause, }), ), attempt, @@ -951,8 +948,6 @@ export function createSubAgentSessionStore( const releaseHandles = (id: string): void => { cancelAskInternal(id, "session handles released"); - // Eviction and prune have no record left to keep honest, so the close - // stays fire-and-forget here; fail() awaits it through detachClose. void detachClose(id)?.catch((err: unknown) => { log.warn("session close during handle release failed: {error}", { error: errorMessage(err), @@ -1262,7 +1257,10 @@ export function createSubAgentSessionStore( // Only the interrupted restore stamps a record of its own; a completed // restore stays a clean lane, as before. if (failLifecycle === "interrupted") { - pendingFailures.set(id, { failure_class: classifySubAgentFailure(err) }); + pendingFailures.set(id, { + failure_class: classifySubAgentFailure(err), + ...(isSubAgentLoopGuardError(err) ? { cause: err.cause } : {}), + }); } endFollowupTurn(id, failLifecycle); pendingFailures.delete(id); @@ -1732,39 +1730,13 @@ export function createSubAgentSessionStore( } }, - async fail( + fail( id: string, error: string, failure?: SubAgentFailureInput, ): Promise { if (failure !== undefined) pendingFailures.set(id, failure); settleCancelsAsks(id, "session failed"); - const current = sessions.get(id); - const failable = - current !== undefined && - isLiveStrip(current.lifecycle) && - current.lifecycle.state !== "cancelled"; - // The record is written once and never amended, so the close it reports - // on has to finish first. A close that rejects or outlives its deadline - // is a cleanup fault, not a released worker. - const closing = failable ? detachClose(id) : undefined; - if (closing !== undefined && failure?.teardownFailed === true) { - // The worker's own bounded teardown already timed out on this agent, - // so the record is partial either way; waiting on a second close - // would only hold the lane and its admission slot for another - // deadline. - void closing.catch((err: unknown) => { - log.warn("backstop close after failed teardown failed: {error}", { - error: errorMessage(err), - }); - }); - } else if (closing !== undefined) { - try { - await closing; - } catch { - releaseFaults.add(id); - } - } mutate(id, (session) => { if ( !isLiveStrip(session.lifecycle) || @@ -1773,7 +1745,8 @@ export function createSubAgentSessionStore( return; // Spawn-path throws already dispose in run.ts's finally; a resumed // persisted agent does not, so the live close handle is the only - // teardown, awaited above. A disposed spawn close is idempotent. + // teardown. Invoke it fire-and-forget (as prune/evict) without + // marking shutdown — a disposed spawn close is idempotent. session.lifecycle = { state: "failed", error }; session.retained = false; session.finishedAt = now(); @@ -1789,6 +1762,7 @@ export function createSubAgentSessionStore( pruneCompleted(); }); pendingFailures.delete(id); + return Promise.resolve(); }, registerCancel(id: string, abort: () => void): void { diff --git a/src/subagent/terminal-failure.test.ts b/src/subagent/terminal-failure.test.ts index 4b8911b60..ba4c8f6e9 100644 --- a/src/subagent/terminal-failure.test.ts +++ b/src/subagent/terminal-failure.test.ts @@ -124,6 +124,60 @@ describe("recoveryFor", () => { ).toEqual({ available: false, reason: "not_retryable" }); }); + test("a loop_guard with a retryable provider cause is eligible", () => { + expect( + recoveryFor({ + failureClass: "loop_guard", + attempt: 1, + followupTurn: false, + cause: providerError("retryable"), + }), + ).toEqual({ available: true, reason: "eligible" }); + expect( + recoveryFor({ + failureClass: "loop_guard", + attempt: 1, + followupTurn: false, + cause: createSubAgentLoopGuardError(providerError("timeout")), + }), + ).toEqual({ available: true, reason: "eligible" }); + }); + + test("a loop_guard without a retryable cause stays not_retryable", () => { + expect( + recoveryFor({ + failureClass: "loop_guard", + attempt: 1, + followupTurn: false, + cause: providerError("credential_failure"), + }), + ).toEqual({ available: false, reason: "not_retryable" }); + expect( + recoveryFor({ + failureClass: "loop_guard", + attempt: 1, + followupTurn: false, + cause: new Error("doom"), + }), + ).toEqual({ available: false, reason: "not_retryable" }); + expect( + recoveryFor({ + failureClass: "loop_guard", + attempt: 1, + followupTurn: true, + cause: providerError("retryable"), + }), + ).toEqual({ available: false, reason: "not_retryable" }); + expect( + recoveryFor({ + failureClass: "loop_guard", + attempt: 2, + followupTurn: false, + cause: providerError("retryable"), + }), + ).toEqual({ available: false, reason: "recovery_exhausted" }); + }); + test("a retryable failure on a follow-up turn is not_retryable", () => { expect( recoveryFor({ diff --git a/src/subagent/terminal-failure.ts b/src/subagent/terminal-failure.ts index 7e6658e73..7aba584af 100644 --- a/src/subagent/terminal-failure.ts +++ b/src/subagent/terminal-failure.ts @@ -68,6 +68,8 @@ export interface SubAgentFailureInput { readonly failure_class: SubAgentFailureClass; /** The worker's own session teardown threw or timed out. */ readonly teardownFailed?: boolean; + /** Doom-loop cause. Read for recovery; omitted from the parent record. */ + readonly cause?: unknown; } export function isProviderFailureClass(cls: SubAgentFailureClass): boolean { @@ -93,7 +95,7 @@ export function createSubAgentLoopGuardError(cause: unknown): Error { return error; } -export function isSubAgentLoopGuardError(err: unknown): boolean { +export function isSubAgentLoopGuardError(err: unknown): err is Error { return err instanceof Error && loopGuardErrors.has(err); } @@ -131,12 +133,28 @@ export function recoveryFor(input: { failureClass: SubAgentFailureClass; attempt: number; followupTurn: boolean; + cause?: unknown; }): SubAgentRecovery { if (input.attempt > 1) { return { available: false, reason: "recovery_exhausted" }; } - if (input.followupTurn || input.failureClass !== "provider_retryable") { + if (input.followupTurn || !isRetryableFailure(input)) { return { available: false, reason: "not_retryable" }; } return { available: true, reason: "eligible" }; } + +function isRetryableFailure(input: { + failureClass: SubAgentFailureClass; + cause?: unknown; +}): boolean { + if (input.failureClass === "provider_retryable") return true; + if (input.failureClass !== "loop_guard") return false; + const cause = isSubAgentLoopGuardError(input.cause) + ? input.cause.cause + : input.cause; + return ( + isResolvedProviderFailureError(cause) && + isRecoverableProviderFailureCategory(cause.category) + ); +}