From 2afa655a5061693ccd57cb9f312e9619e902f06c Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 31 Aug 2026 15:20:03 -0700 Subject: [PATCH 1/7] Store one worker lifecycle instead of two status fields Snapshot status and verb lifecycle are now projections of one stored union. fail() is a first-class failed state, not shutdown, so wait and resume can tell a throw from a close. Pin/unpin keep uncollected wait results off the display-cap prune. --- src/subagent/lifecycle.test.ts | 68 ++++++ src/subagent/lifecycle.ts | 77 +++++++ src/subagent/session-store.test.ts | 129 +++++++++++ src/subagent/session-store.ts | 269 ++++++++++++++-------- src/tui/runner-host.test.ts | 1 + tests/unit/subagent-session-store.test.ts | 56 +++++ 6 files changed, 498 insertions(+), 102 deletions(-) create mode 100644 src/subagent/lifecycle.test.ts create mode 100644 src/subagent/lifecycle.ts diff --git a/src/subagent/lifecycle.test.ts b/src/subagent/lifecycle.test.ts new file mode 100644 index 00000000..dd6ca8b1 --- /dev/null +++ b/src/subagent/lifecycle.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, test } from "bun:test"; + +import { + isAlreadyClosed, + isLiveStrip, + isResumableLifecycle, + projectLifecycleStatus, + projectStripStatus, + type StripStatus, + type WorkerLifecycle, +} from "./lifecycle.js"; + +describe("WorkerLifecycle projections", () => { + test("strip status maps each stored state", () => { + const cases: [WorkerLifecycle, StripStatus][] = [ + [{ state: "pending_init" }, "running"], + [{ state: "running" }, "running"], + [{ state: "interrupted" }, "running"], + [{ state: "completed", report: "ok" }, "done"], + [{ state: "failed", error: "boom" }, "failed"], + [{ state: "cancelled", error: "stop" }, "cancelled"], + [{ state: "shutdown" }, "cancelled"], + ]; + for (const [lifecycle, status] of cases) { + expect(projectStripStatus(lifecycle)).toBe(status); + } + }); + + test("verb lifecycleStatus does not leak cancelled or failed", () => { + expect(projectLifecycleStatus({ state: "pending_init" })).toBe("pending_init"); + expect(projectLifecycleStatus({ state: "running" })).toBe("running"); + expect(projectLifecycleStatus({ state: "completed", report: "ok" })).toBe("completed"); + expect(projectLifecycleStatus({ state: "interrupted" })).toBe("interrupted"); + expect(projectLifecycleStatus({ state: "cancelled" })).toBe("interrupted"); + expect(projectLifecycleStatus({ state: "failed", error: "boom" })).toBe("shutdown"); + expect(projectLifecycleStatus({ state: "shutdown" })).toBe("shutdown"); + }); + + test("resume gate is retained completed or interrupted only", () => { + expect(isResumableLifecycle(true, { state: "completed", report: "ok" })).toBe(true); + expect(isResumableLifecycle(true, { state: "interrupted" })).toBe(true); + expect(isResumableLifecycle(true, { state: "cancelled" })).toBe(false); + expect(isResumableLifecycle(true, { state: "failed", error: "x" })).toBe(false); + expect(isResumableLifecycle(true, { state: "shutdown" })).toBe(false); + expect(isResumableLifecycle(false, { state: "completed", report: "ok" })).toBe(false); + expect(isResumableLifecycle(undefined, { state: "interrupted" })).toBe(false); + }); + + test("live strip is pending_init, running, and interrupted", () => { + expect(isLiveStrip({ state: "pending_init" })).toBe(true); + expect(isLiveStrip({ state: "running" })).toBe(true); + expect(isLiveStrip({ state: "interrupted" })).toBe(true); + expect(isLiveStrip({ state: "completed", report: "ok" })).toBe(false); + expect(isLiveStrip({ state: "cancelled" })).toBe(false); + expect(isLiveStrip({ state: "failed", error: "x" })).toBe(false); + expect(isLiveStrip({ state: "shutdown" })).toBe(false); + }); + + test("already-closed is failed or shutdown so close_agent does not wait", () => { + expect(isAlreadyClosed({ state: "failed", error: "x" })).toBe(true); + expect(isAlreadyClosed({ state: "shutdown" })).toBe(true); + expect(isAlreadyClosed({ state: "pending_init" })).toBe(false); + expect(isAlreadyClosed({ state: "running" })).toBe(false); + expect(isAlreadyClosed({ state: "interrupted" })).toBe(false); + expect(isAlreadyClosed({ state: "completed", report: "ok" })).toBe(false); + expect(isAlreadyClosed({ state: "cancelled" })).toBe(false); + }); +}); diff --git a/src/subagent/lifecycle.ts b/src/subagent/lifecycle.ts new file mode 100644 index 00000000..b7c7daf4 --- /dev/null +++ b/src/subagent/lifecycle.ts @@ -0,0 +1,77 @@ +/** + * Stored worker lifecycle for SubAgentSessionStore. + * + * `status` / `lifecycleStatus` on snapshots are projections of this union. + * `not_found` is a query result only and is never stored. + */ + +export type WorkerLifecycle = + | { state: "pending_init" } + | { state: "running" } + | { state: "completed"; report: string } + | { state: "failed"; error: string } + | { state: "interrupted"; report?: string } + | { state: "cancelled"; report?: string; error?: string } + | { state: "shutdown"; report?: string; error?: string }; + +export type StripStatus = "running" | "done" | "failed" | "cancelled"; + +export type VerbLifecycleStatus = + "pending_init" | "running" | "interrupted" | "completed" | "shutdown"; + +/** TUI / Agents-strip status. Interrupted lingers as running. */ +export function projectStripStatus(lifecycle: WorkerLifecycle): StripStatus { + switch (lifecycle.state) { + case "pending_init": + case "running": + case "interrupted": + return "running"; + case "completed": + return "done"; + case "failed": + return "failed"; + case "cancelled": + case "shutdown": + return "cancelled"; + } +} + +/** + * Verb JSON (close/resume/interrupt). Does not leak `cancelled` or `failed`: + * cancelled → interrupted, failed → shutdown. + */ +export function projectLifecycleStatus(lifecycle: WorkerLifecycle): VerbLifecycleStatus { + switch (lifecycle.state) { + case "pending_init": + return "pending_init"; + case "running": + return "running"; + case "completed": + return "completed"; + case "interrupted": + case "cancelled": + return "interrupted"; + case "failed": + case "shutdown": + return "shutdown"; + } +} + +/** pending_init / running / interrupted — strip still shows running. */ +export function isLiveStrip(lifecycle: WorkerLifecycle): boolean { + return projectStripStatus(lifecycle) === "running"; +} + +/** Agent already disposed — close_agent must not wait for a handle. */ +export function isAlreadyClosed(lifecycle: WorkerLifecycle): boolean { + return lifecycle.state === "failed" || lifecycle.state === "shutdown"; +} + +export function isResumableLifecycle( + retained: boolean | undefined, + lifecycle: WorkerLifecycle, +): boolean { + return ( + retained === true && (lifecycle.state === "completed" || lifecycle.state === "interrupted") + ); +} diff --git a/src/subagent/session-store.test.ts b/src/subagent/session-store.test.ts index e0d4c325..8fe6f315 100644 --- a/src/subagent/session-store.test.ts +++ b/src/subagent/session-store.test.ts @@ -754,3 +754,132 @@ describe("interrupt stamps finishedAt once", () => { expect(fleetProgress(terminal, t).running).toBe(0); }); }); + +describe("CL-7269 one stored worker lifecycle", () => { + test("fail() stores failed, projects strip failed and verb shutdown", () => { + const store = createSubAgentSessionStore(); + const session = store.start({ description: "d", agentId: "a", brief: "b" }); + store.fail(session.id, "provider 500"); + const after = store.get(session.id); + expect(after?.status).toBe("failed"); + expect(after?.lifecycle.state).toBe("failed"); + expect(after?.lifecycleStatus).toBe("shutdown"); + expect(after?.error).toBe("provider 500"); + }); + + test("cancel() is not resumable; interruptOne() is when retained", () => { + const store = createSubAgentSessionStore(); + const cancelled = store.start({ + description: "c", + agentId: "a", + brief: "b", + retained: true, + }); + store.markRunning(cancelled.id); + store.registerFollowup(cancelled.id, async () => "nope"); + expect(store.cancel(cancelled.id, "operator kill")).toBe(true); + const afterCancel = store.get(cancelled.id); + expect(afterCancel?.status).toBe("cancelled"); + expect(afterCancel?.lifecycle.state).toBe("cancelled"); + expect(afterCancel?.lifecycleStatus).toBe("interrupted"); + expect(afterCancel?.retained).toBe(false); + expect(store.resumeOne(cancelled.id, "continue").ok).toBe(false); + + const interrupted = store.start({ + description: "i", + agentId: "a", + brief: "b", + retained: true, + }); + store.markRunning(interrupted.id); + store.registerInterrupt(interrupted.id, () => {}); + store.registerFollowup(interrupted.id, async () => "next"); + expect(store.interruptOne(interrupted.id).ok).toBe(true); + const afterInterrupt = store.get(interrupted.id); + expect(afterInterrupt?.lifecycle.state).toBe("interrupted"); + expect(afterInterrupt?.status).toBe("running"); + expect(afterInterrupt?.lifecycleStatus).toBe("interrupted"); + expect(afterInterrupt?.retained).toBe(true); + expect(store.resumeOne(interrupted.id, "continue")).toEqual({ ok: true, status: "running" }); + }); + + test("complete() after cancel() no-ops", () => { + const store = createSubAgentSessionStore(); + const session = store.start({ description: "d", agentId: "a", brief: "b" }); + store.cancel(session.id, "operator kill"); + store.complete(session.id, "should not win"); + const after = store.get(session.id); + expect(after?.status).toBe("cancelled"); + expect(after?.lifecycle.state).toBe("cancelled"); + expect(after?.report).toBeUndefined(); + }); + + test("pin keeps a session past maxCompleted; unpin allows prune", () => { + let n = 0; + let t = 0; + const store = createSubAgentSessionStore({ + maxCompleted: 1, + createId: () => `s-${++n}`, + now: () => ++t, + }); + const pinned = store.start({ description: "keep", agentId: "a", brief: "b" }); + store.pin(pinned.id); + store.complete(pinned.id, "report keep"); + + const extra1 = store.start({ description: "drop-me", agentId: "a", brief: "b" }); + store.complete(extra1.id, "report extra"); + expect(store.get(pinned.id)?.id).toBe(pinned.id); + expect(store.get(extra1.id)?.id).toBe(extra1.id); + + const extra2 = store.start({ description: "also", agentId: "a", brief: "b" }); + store.complete(extra2.id, "report also"); + expect(store.get(pinned.id)).toBeDefined(); + expect(store.get(extra1.id)).toBeUndefined(); + expect(store.get(extra2.id)).toBeDefined(); + + store.unpin(pinned.id); + const extra3 = store.start({ description: "prune-pinned", agentId: "a", brief: "b" }); + store.complete(extra3.id, "report prune"); + expect(store.get(pinned.id)).toBeUndefined(); + expect(store.get(extra3.id)).toBeDefined(); + }); + + test("closeOne after fail() returns immediately and leaves stored failed", async () => { + const store = createSubAgentSessionStore(); + const session = store.start({ description: "d", agentId: "a", brief: "b" }); + store.fail(session.id, "provider 500"); + const started = Date.now(); + const status = await store.closeOne(session.id, 5000); + expect(Date.now() - started).toBeLessThan(200); + expect(status).toBe("shutdown"); + expect(store.get(session.id)?.lifecycle.state).toBe("failed"); + }); + + test("closeOne during setup then fail() does not wait the deadline", async () => { + const store = createSubAgentSessionStore(); + const session = store.start({ description: "d", agentId: "a", brief: "b" }); + const started = Date.now(); + const closePromise = store.closeOne(session.id, 5000); + setTimeout(() => store.fail(session.id, "boom"), 15); + expect(await closePromise).toBe("shutdown"); + expect(Date.now() - started).toBeLessThan(200); + expect(store.get(session.id)?.lifecycle.state).toBe("failed"); + }); + + test("start() reuse of an id drops leftover pins", () => { + let t = 0; + const store = createSubAgentSessionStore({ + maxCompleted: 1, + now: () => ++t, + }); + store.start({ id: "reuse", description: "old", agentId: "a", brief: "b" }); + store.pin("reuse"); + store.start({ id: "reuse", description: "new", agentId: "a", brief: "b" }); + store.complete("reuse", "new report"); + const extra = store.start({ description: "other", agentId: "a", brief: "b" }); + store.complete(extra.id, "other report"); + const extra2 = store.start({ description: "prune", agentId: "a", brief: "b" }); + store.complete(extra2.id, "prune report"); + expect(store.get("reuse")).toBeUndefined(); + }); +}); diff --git a/src/subagent/session-store.ts b/src/subagent/session-store.ts index c1a47d76..aa0235a0 100644 --- a/src/subagent/session-store.ts +++ b/src/subagent/session-store.ts @@ -8,6 +8,14 @@ import type { ReactorEmittedEvent } from "@intx/inference"; import { getLogger } from "@intx/log"; import { LOG_NAMESPACE_ROOT } from "../branding.js"; import { DEFAULT_CLOSE_DEADLINE_MS } from "./dispose.js"; +import { + isAlreadyClosed, + isLiveStrip, + isResumableLifecycle, + projectLifecycleStatus, + projectStripStatus, + type WorkerLifecycle, +} from "./lifecycle.js"; import type { ForcedStopReason } from "./stop-policy.js"; import { toolCallPreview } from "./tool-preview.js"; @@ -16,16 +24,17 @@ const log = getLogger([LOG_NAMESPACE_ROOT, "subagent", "session-store"]); export type SubAgentSessionStatus = "running" | "done" | "failed" | "cancelled"; /** - * CL-6943: lifecycle status surfaced to the parent for the reusable-session - * verbs (close_agent / resume_agent), independent of `SubAgentSessionStatus` - * above (which is the older TUI-transcript status and is left alone here). - * `interrupted` is not produced by anything in this lane — `cancel()` sets - * it today (the pre-existing operator-cancel path), and the interrupt_agent - * lane lands later reusing the same value, not a new one. + * Lifecycle status surfaced to the parent for the reusable-session verbs + * (close_agent / resume_agent). Snapshot `lifecycleStatus` is a projection of + * stored `WorkerLifecycle` and never leaks `cancelled` or `failed` + * (cancelled → interrupted, failed → shutdown). `not_found` is a query result + * only — never stored. */ export type AgentLifecycleStatus = "pending_init" | "running" | "interrupted" | "completed" | "shutdown" | "not_found"; +export type { WorkerLifecycle }; + // Compact transcript entries suitable for TUI render without depending on the // TUI ContentBlock type (keeps subagent free of a reverse dependency on tui/). export type SubAgentTranscriptEntry = @@ -53,7 +62,10 @@ export interface SubAgentSession { description: string; agentId: string; brief: string; + /** Projection of `lifecycle` for TUI / Agents strip. */ status: SubAgentSessionStatus; + /** Stored source of truth. Snapshot copies it; do not mutate independently. */ + lifecycle: WorkerLifecycle; toolNames: string[]; // Name, preview, and start clock of the OLDEST outstanding call — the one // that explains the longest silence. All three are derived from @@ -94,8 +106,10 @@ export interface SubAgentSession { // a nested (one-hop) dispatch. Undefined for top-level sessions started // directly from the primary session's task tool. parentSessionId?: string; - // CL-6943: lifecycle status for the reusable-session verbs. Defaults to - // "pending_init" until the run wires up markRunning(); see the type doc. + /** + * Projection of `lifecycle` for close/resume/interrupt JSON. Maps cancelled → + * interrupted and failed → shutdown so those verbs do not leak new enum values. + */ lifecycleStatus: AgentLifecycleStatus; // True when this session's agent is meant to survive a clean completion // (spawn_agent opts in). An open retained session ("completed" or @@ -214,6 +228,12 @@ export interface SubAgentSessionStore { message: string, opts?: { interrupt?: boolean; onFollowupReply?: (reply: string) => void }, ): { ok: true; status: AgentLifecycleStatus } | { ok: false; status: AgentLifecycleStatus }; + /** + * Refcount so wait mailboxes can pin an uncollected result. pruneCompleted + * will not delete a session while its pin count is greater than zero. + */ + pin(id: string): void; + unpin(id: string): void; subscribe(listener: () => void): () => void; clear(): void; } @@ -248,6 +268,9 @@ interface EvictedRecord { hint: string; } +/** In-map record: `status` / `lifecycleStatus` exist only on snapshots. */ +type StoredSession = Omit; + let nextId = 0; function defaultCreateId(): string { nextId += 1; @@ -259,7 +282,7 @@ function defaultCreateId(): string { * be shown beside another call's clock. Called after every change to * `outstandingTools`. */ -function syncCurrentTool(session: SubAgentSession): void { +function syncCurrentTool(session: StoredSession): void { let oldest: OutstandingToolCall | undefined; for (const call of session.outstandingTools) { if (oldest === undefined || call.startedAt < oldest.startedAt) oldest = call; @@ -276,7 +299,7 @@ function syncCurrentTool(session: SubAgentSession): void { * transcript stores. */ function beginToolCall( - session: SubAgentSession, + session: StoredSession, callId: string, name: string, nowMs: number, @@ -303,7 +326,7 @@ function beginToolCall( /** Refresh the outstanding call's preview once more of its arguments stream in. */ function refreshToolPreview( - session: SubAgentSession, + session: StoredSession, callId: string, name: string, rawArgs: string, @@ -318,14 +341,14 @@ function refreshToolPreview( * Retires exactly the call that finished. A result carrying an id we never saw * start retires nothing, rather than silently clearing a live sibling's clock. */ -function endToolCall(session: SubAgentSession, callId: string): void { +function endToolCall(session: StoredSession, callId: string): void { const index = session.outstandingTools.findIndex((c) => c.callId === callId); if (index === -1) return; session.outstandingTools.splice(index, 1); syncCurrentTool(session); } -function clearToolCalls(session: SubAgentSession): void { +function clearToolCalls(session: StoredSession): void { session.outstandingTools.length = 0; syncCurrentTool(session); } @@ -361,7 +384,9 @@ export function createSubAgentSessionStore( const createId = options.createId ?? defaultCreateId; // Insertion order: older first. list() returns a snapshot in that order. - const sessions = new Map(); + const sessions = new Map(); + // Pin refcount: wait mailboxes hold a pin until they collect the result. + const pinCounts = new Map(); // Live abort hooks keyed by session id. Cleared on terminal transition. const cancelHandles = new Map void>(); // CL-6943: bounded close functions keyed by session id, for close_agent. @@ -381,9 +406,9 @@ export function createSubAgentSessionStore( // session evicted purely to bound retention memory. const evicted = new Map(); - const recordEviction = (session: SubAgentSession): void => { + const recordEviction = (session: StoredSession): void => { evicted.set(session.id, { - lifecycleStatus: session.lifecycleStatus, + lifecycleStatus: projectLifecycleStatus(session.lifecycle), hint: EVICTED_RETENTION_HINT, }); if (evicted.size > MAX_EVICTED_TOMBSTONES) { @@ -409,7 +434,7 @@ export function createSubAgentSessionStore( snapshotCache.delete(id); }; - const snapshotOf = (session: SubAgentSession): SubAgentSession => { + const snapshotOf = (session: StoredSession): SubAgentSession => { const revision = revisions.get(session.id) ?? 0; const cached = snapshotCache.get(session.id); if (cached !== undefined && cached.revision === revision) return cached.snapshot; @@ -422,9 +447,9 @@ export function createSubAgentSessionStore( for (const listener of listeners) listener(); }; - const markCancelled = (session: SubAgentSession, reason: string): void => { - session.status = "cancelled"; - session.lifecycleStatus = "interrupted"; + const markCancelled = (session: StoredSession, reason: string): void => { + session.lifecycle = { state: "cancelled", error: reason }; + session.retained = false; session.finishedAt = now(); session.lastActivityAt = now(); clearToolCalls(session); @@ -443,7 +468,7 @@ export function createSubAgentSessionStore( const cancelSession = (id: string, reason: string): boolean => { const session = sessions.get(id); - if (session === undefined || session.status !== "running") return false; + if (session === undefined || !isLiveStrip(session.lifecycle)) return false; const abort = cancelHandles.get(id); // Flip status first so concurrent complete/fail see a non-running session, // then fire the abort handle (which may re-enter via signal listeners). @@ -459,7 +484,7 @@ export function createSubAgentSessionStore( return true; }; - const pushEntry = (session: SubAgentSession, entry: SubAgentTranscriptEntry): void => { + const pushEntry = (session: StoredSession, entry: SubAgentTranscriptEntry): void => { session.entries.push(entry); if (session.entries.length > maxEntries) { session.entries.splice(0, session.entries.length - maxEntries); @@ -490,9 +515,14 @@ export function createSubAgentSessionStore( // An open retained session (spawn_agent's reusable-session contract: // retained:true and still addressable — "completed" or "interrupted") is // governed by pruneRetained's own cap below, not this one. - const isOpenRetained = (s: SubAgentSession): boolean => + const isOpenRetained = (s: StoredSession): boolean => s.retained === true && - (s.lifecycleStatus === "completed" || s.lifecycleStatus === "interrupted"); + (s.lifecycle.state === "completed" || s.lifecycle.state === "interrupted"); + + const isPinned = (id: string): boolean => (pinCounts.get(id) ?? 0) > 0; + + const isPrunableCompleted = (s: StoredSession): boolean => + !isLiveStrip(s.lifecycle) && !isOpenRetained(s) && !isPinned(s.id); // CL-7001/CL-7007: `maxCompleted` bounds every ordinary finished session — // one that was never retained, or a retained one already closed via @@ -500,12 +530,13 @@ export function createSubAgentSessionStore( // cap and was never sized to also be the retention policy for reusable // sessions; open retained sessions are excluded here and bounded instead // by pruneRetained. A session that was resumed and is actively running - // again (lifecycleStatus "running") is still excluded: it has a live - // caller, not an idle leak. + // again (lifecycle state "running") is still excluded: it has a live + // caller, not an idle leak. Pinned ids (uncollected wait results) are + // also excluded so maxCompleted cannot delete them. const pruneCompleted = (): void => { if (maxCompleted <= 0) { for (const [id, s] of sessions) { - if (s.status !== "running" && s.lifecycleStatus !== "running" && !isOpenRetained(s)) { + if (isPrunableCompleted(s)) { releaseHandles(id); sessions.delete(id); forgetRevision(id); @@ -514,9 +545,7 @@ export function createSubAgentSessionStore( return; } const finished = [...sessions.values()] - .filter( - (s) => s.status !== "running" && s.lifecycleStatus !== "running" && !isOpenRetained(s), - ) + .filter(isPrunableCompleted) .sort((a, b) => (a.finishedAt ?? 0) - (b.finishedAt ?? 0)); const excess = finished.length - maxCompleted; if (excess <= 0) return; @@ -575,7 +604,7 @@ export function createSubAgentSessionStore( }; const check = (): void => { const session = sessions.get(id); - if (session === undefined || session.lifecycleStatus === "shutdown") { + if (session === undefined || isAlreadyClosed(session.lifecycle)) { finish(undefined); return; } @@ -588,7 +617,7 @@ export function createSubAgentSessionStore( }); }; - const mutate = (id: string, fn: (session: SubAgentSession) => void): void => { + const mutate = (id: string, fn: (session: StoredSession) => void): void => { const session = sessions.get(id); if (session === undefined) return; fn(session); @@ -605,26 +634,31 @@ export function createSubAgentSessionStore( // that restore — do not rewrite interrupted back to completed. const beginFollowupTurn = (id: string): void => { mutate(id, (s) => { - s.status = "running"; - s.lifecycleStatus = "running"; + s.lifecycle = { state: "running" }; delete s.finishedAt; }); }; - const endFollowupTurn = (id: string, lifecycleStatus: AgentLifecycleStatus): void => { + const endFollowupTurn = (id: string, restore: "completed" | "interrupted"): void => { mutate(id, (s) => { - if (s.lifecycleStatus === "interrupted") { + if (s.lifecycle.state !== "running" && s.lifecycle.state !== "pending_init") { s.finishedAt = s.finishedAt ?? now(); return; } - s.lifecycleStatus = lifecycleStatus; + if (restore === "interrupted") { + s.lifecycle = { + state: "interrupted", + ...(s.report !== undefined ? { report: s.report } : {}), + }; + } else { + s.lifecycle = { state: "completed", report: s.report ?? "" }; + } s.finishedAt = now(); - s.status = lifecycleStatus === "interrupted" ? "running" : "done"; }); }; const queueFollowupTurn = ( id: string, message: string, - failLifecycle: AgentLifecycleStatus, + failLifecycle: "completed" | "interrupted", opts?: { onReply?: (reply: string) => void; onFail?: (error: unknown) => void }, ): void => { const followup = followupHandles.get(id); @@ -633,10 +667,16 @@ export function createSubAgentSessionStore( void followup(message) .then((reply) => { const still = sessions.get(id); - if (still === undefined || still.lifecycleStatus === "shutdown") return; + if (still === undefined) return; + if ( + still.lifecycle.state === "shutdown" || + still.lifecycle.state === "cancelled" || + still.lifecycle.state === "failed" + ) { + return; + } mutate(id, (s) => { - s.status = "done"; - s.lifecycleStatus = "completed"; + s.lifecycle = { state: "completed", report: reply }; s.finishedAt = now(); s.report = reply; pushEntry(s, { kind: "report", content: capText(reply, maxEntryChars) }); @@ -682,13 +722,14 @@ export function createSubAgentSessionStore( interruptHandles.delete(id); followupHandles.delete(id); deliverHandles.delete(id); + pinCounts.delete(id); forgetRevision(id); - const session: SubAgentSession = { + const session: StoredSession = { id, description: input.description, agentId: input.agentId, brief: input.brief, - status: "running", + lifecycle: { state: "pending_init" }, toolNames: [], currentToolName: null, currentToolPreview: null, @@ -697,7 +738,6 @@ export function createSubAgentSessionStore( entries: [], startedAt: now(), lastActivityAt: now(), - lifecycleStatus: "pending_init", ...(input.retained === true ? { retained: true } : {}), ...(input.parentSessionId !== undefined ? { parentSessionId: input.parentSessionId } : {}), }; @@ -709,7 +749,7 @@ export function createSubAgentSessionStore( appendEvent(id: string, event: ReactorEmittedEvent): void { mutate(id, (session) => { - if (session.status !== "running") return; + if (!isLiveStrip(session.lifecycle)) return; switch (event.type) { case "inference.text.delta": { const token = (event.data as { token?: unknown })?.token; @@ -867,17 +907,10 @@ export function createSubAgentSessionStore( mutate(id, (session) => { // Cancel wins races: a late complete after operator cancel must not // resurrect the session as done. - if (session.status !== "running") return; - session.status = "done"; - // CL-6943: retained sessions stay "completed" (open, reusable) here — - // only close_agent (closeOne) moves them to "shutdown". - session.lifecycleStatus = "completed"; - // CL-7001: a disposed salvage (deadline/cancel) resolves through - // this same path but run.ts has already torn its agent down — clear - // `retained` so resumeOne's `retained === true` gate can never see - // it as open, without touching the lifecycleStatus invariant every - // other completion (including a never-retained one) already relies - // on. + if (!isLiveStrip(session.lifecycle) || session.lifecycle.state === "cancelled") return; + // Interrupted is still strip-live; a settling run may complete. Operator + // cancel is not live for this path because state is cancelled. + session.lifecycle = { state: "completed", report }; if (!agentRetained) session.retained = false; session.finishedAt = now(); clearToolCalls(session); @@ -895,8 +928,7 @@ export function createSubAgentSessionStore( fail(id: string, error: string): void { mutate(id, (session) => { - if (session.status !== "running") return; - session.status = "failed"; + if (!isLiveStrip(session.lifecycle) || session.lifecycle.state === "cancelled") return; // A thrown run always tears down its agent in run.ts's finally // (persist only skips teardown on a clean success) — so there is // nothing left to resume here, and retained no longer applies. @@ -905,7 +937,7 @@ export function createSubAgentSessionStore( // it never reaches this function. complete() carries the equivalent // "agent was actually disposed" check for that case via its // agentRetained flag; this function only ever needed to cover throws. - session.lifecycleStatus = "shutdown"; + session.lifecycle = { state: "failed", error }; session.retained = false; session.finishedAt = now(); clearToolCalls(session); @@ -922,13 +954,13 @@ export function createSubAgentSessionStore( registerCancel(id: string, abort: () => void): void { const session = sessions.get(id); - if (session === undefined || session.status !== "running") return; + if (session === undefined || !isLiveStrip(session.lifecycle)) return; cancelHandles.set(id, abort); }, markRunning(id: string): void { mutate(id, (session) => { - if (session.lifecycleStatus === "pending_init") session.lifecycleStatus = "running"; + if (session.lifecycle.state === "pending_init") session.lifecycle = { state: "running" }; }); }, @@ -951,7 +983,7 @@ export function createSubAgentSessionStore( if (evicted.has(id)) return "shutdown"; return "not_found"; } - if (session.lifecycleStatus === "shutdown") return "shutdown"; + if (isAlreadyClosed(session.lifecycle)) return projectLifecycleStatus(session.lifecycle); let close = closeHandles.get(id); if (close === undefined) { // CL-7001: close_agent landed in the setup window — the session @@ -960,16 +992,18 @@ export function createSubAgentSessionStore( // returning "shutdown" immediately: that used to report false // success while leaving the eventual agent unreleasable forever // (the early return above short-circuits every retry once - // lifecycleStatus flips). + // lifecycle flips). close = await waitForCloseHandle(id, deadlineMs); const stillHere = sessions.get(id); if (stillHere === undefined) return "not_found"; - if (stillHere.lifecycleStatus === "shutdown") return "shutdown"; + if (isAlreadyClosed(stillHere.lifecycle)) { + return projectLifecycleStatus(stillHere.lifecycle); + } if (close === undefined) { // Never became closeable within the deadline: report the honest // in-progress status rather than a false "shutdown" — the caller // can retry, and this session is still findable to retry against. - return stillHere.lifecycleStatus; + return projectLifecycleStatus(stillHere.lifecycle); } } closeHandles.delete(id); @@ -985,13 +1019,18 @@ export function createSubAgentSessionStore( new Promise((resolve) => setTimeout(resolve, deadlineMs)), ]); mutate(id, (s) => { - s.lifecycleStatus = "shutdown"; - s.retained = false; - if (s.status === "running") { - s.status = "cancelled"; + const wasLive = isLiveStrip(s.lifecycle); + const error = s.error ?? (wasLive ? "Closed by close_agent" : undefined); + if (wasLive) { s.finishedAt = s.finishedAt ?? now(); - s.error = s.error ?? "Closed by close_agent"; + if (error !== undefined) s.error = error; } + s.lifecycle = { + state: "shutdown", + ...(s.report !== undefined ? { report: s.report } : {}), + ...(error !== undefined ? { error } : {}), + }; + s.retained = false; }); cancelHandles.delete(id); interruptHandles.delete(id); @@ -1023,15 +1062,15 @@ export function createSubAgentSessionStore( ): { ok: true; status: AgentLifecycleStatus } | { ok: false; status: AgentLifecycleStatus } { const session = sessions.get(id); if (session === undefined) return { ok: false, status: "not_found" }; - if (session.status !== "running" || session.lifecycleStatus !== "running") { - return { ok: false, status: session.lifecycleStatus }; + if (session.lifecycle.state !== "running") { + return { ok: false, status: projectLifecycleStatus(session.lifecycle) }; } if (opts?.interrupt === true) { const interrupt = interruptHandles.get(id); const followup = followupHandles.get(id); if (interrupt === undefined || followup === undefined) { - return { ok: false, status: session.lifecycleStatus }; + return { ok: false, status: projectLifecycleStatus(session.lifecycle) }; } interrupt(); queueFollowupTurn(id, message, "interrupted", { @@ -1042,7 +1081,9 @@ export function createSubAgentSessionStore( } const deliver = deliverHandles.get(id); - if (deliver === undefined) return { ok: false, status: session.lifecycleStatus }; + if (deliver === undefined) { + return { ok: false, status: projectLifecycleStatus(session.lifecycle) }; + } deliver(message); return { ok: true, status: "running" }; }, @@ -1050,12 +1091,19 @@ export function createSubAgentSessionStore( interruptOne(id: string): { ok: true } | { ok: false; status: AgentLifecycleStatus } { const session = sessions.get(id); if (session === undefined) return { ok: false, status: "not_found" }; - if (session.status !== "running") return { ok: false, status: session.lifecycleStatus }; + if (!isLiveStrip(session.lifecycle)) { + return { ok: false, status: projectLifecycleStatus(session.lifecycle) }; + } const interrupt = interruptHandles.get(id); - if (interrupt === undefined) return { ok: false, status: session.lifecycleStatus }; + if (interrupt === undefined) { + return { ok: false, status: projectLifecycleStatus(session.lifecycle) }; + } interrupt(); mutate(id, (s) => { - s.lifecycleStatus = "interrupted"; + s.lifecycle = { + state: "interrupted", + ...(s.report !== undefined ? { report: s.report } : {}), + }; s.finishedAt = s.finishedAt ?? now(); }); pruneRetained(); @@ -1080,15 +1128,15 @@ export function createSubAgentSessionStore( } return { ok: false, status: "not_found" }; } - if ( - session.retained !== true || - (session.lifecycleStatus !== "completed" && session.lifecycleStatus !== "interrupted") - ) { - return { ok: false, status: session.lifecycleStatus }; + if (!isResumableLifecycle(session.retained, session.lifecycle)) { + return { ok: false, status: projectLifecycleStatus(session.lifecycle) }; } const followup = followupHandles.get(id); - if (followup === undefined) return { ok: false, status: session.lifecycleStatus }; - const priorLifecycle = session.lifecycleStatus; + if (followup === undefined) { + return { ok: false, status: projectLifecycleStatus(session.lifecycle) }; + } + const priorLifecycle = + session.lifecycle.state === "interrupted" ? "interrupted" : "completed"; queueFollowupTurn(id, message, priorLifecycle, { ...(opts?.onReply !== undefined ? { onReply: opts.onReply } : {}), ...(opts?.onFail !== undefined ? { onFail: opts.onFail } : {}), @@ -1103,28 +1151,43 @@ export function createSubAgentSessionStore( }, cancelAll(reason = DEFAULT_CANCEL_REASON): string[] { - const running = [...sessions.values()].filter((s) => s.status === "running"); + // Snapshot before cancelSession: markCancelled clears retained, and a + // resumed retained worker is strip-live so the first loop would otherwise + // skip the close-handle pass (CL-7001). + const retainedIds = [...sessions.values()] + .filter((s) => s.retained === true && s.lifecycle.state !== "shutdown") + .map((s) => s.id); + const running = [...sessions.values()].filter((s) => isLiveStrip(s.lifecycle)); const cancelled: string[] = []; for (const session of running) { if (cancelSession(session.id, reason)) cancelled.push(session.id); } - // CL-7001: a retained session is "done", not "running", so the loop - // above always skipped it — both /clear and session-close route - // through cancelAll, so a retained worker's LSP sidecars, reactor, and - // heldLocks entry outlived the parent turn indefinitely. Release every - // still-open retained session here too, regardless of `status`. - for (const session of sessions.values()) { - if (session.retained === true && session.lifecycleStatus !== "shutdown") { - releaseHandles(session.id); - mutate(session.id, (s) => { - s.lifecycleStatus = "shutdown"; - s.retained = false; - }); - } + for (const id of retainedIds) { + const session = sessions.get(id); + if (session === undefined || session.lifecycle.state === "shutdown") continue; + releaseHandles(id); + mutate(id, (s) => { + s.lifecycle = { + state: "shutdown", + ...(s.report !== undefined ? { report: s.report } : {}), + ...(s.error !== undefined ? { error: s.error } : {}), + }; + s.retained = false; + }); } return cancelled; }, + pin(id: string): void { + pinCounts.set(id, (pinCounts.get(id) ?? 0) + 1); + }, + + unpin(id: string): void { + const next = (pinCounts.get(id) ?? 0) - 1; + if (next <= 0) pinCounts.delete(id); + else pinCounts.set(id, next); + }, + subscribe(listener: () => void): () => void { listeners.add(listener); return () => { @@ -1143,6 +1206,7 @@ export function createSubAgentSessionStore( followupHandles.clear(); deliverHandles.clear(); sessions.clear(); + pinCounts.clear(); revisions.clear(); snapshotCache.clear(); evicted.clear(); @@ -1151,13 +1215,14 @@ export function createSubAgentSessionStore( }; } -function cloneSession(session: SubAgentSession): SubAgentSession { +function cloneSession(session: StoredSession): SubAgentSession { return { id: session.id, description: session.description, agentId: session.agentId, brief: session.brief, - status: session.status, + status: projectStripStatus(session.lifecycle), + lifecycle: { ...session.lifecycle }, toolNames: [...session.toolNames], currentToolName: session.currentToolName, currentToolPreview: session.currentToolPreview, @@ -1166,7 +1231,7 @@ function cloneSession(session: SubAgentSession): SubAgentSession { entries: session.entries.map(cloneEntry), startedAt: session.startedAt, lastActivityAt: session.lastActivityAt, - lifecycleStatus: session.lifecycleStatus, + lifecycleStatus: projectLifecycleStatus(session.lifecycle), ...(session.retained !== undefined ? { retained: session.retained } : {}), ...(session.finishedAt !== undefined ? { finishedAt: session.finishedAt } : {}), ...(session.report !== undefined ? { report: session.report } : {}), diff --git a/src/tui/runner-host.test.ts b/src/tui/runner-host.test.ts index aad677ce..e630039a 100644 --- a/src/tui/runner-host.test.ts +++ b/src/tui/runner-host.test.ts @@ -52,6 +52,7 @@ function session(over: Partial): SubAgentSession { agentId: "explorer", brief: "", status: "running", + lifecycle: { state: "running" }, toolNames: [], currentToolName: null, currentToolPreview: null, diff --git a/tests/unit/subagent-session-store.test.ts b/tests/unit/subagent-session-store.test.ts index beede85f..3c9cefaa 100644 --- a/tests/unit/subagent-session-store.test.ts +++ b/tests/unit/subagent-session-store.test.ts @@ -138,6 +138,8 @@ describe("createSubAgentSessionStore", () => { store.fail("s-3", "provider 500"); const session = store.get("s-3"); expect(session?.status).toBe("failed"); + expect(session?.lifecycle.state).toBe("failed"); + expect(session?.lifecycleStatus).toBe("shutdown"); expect(session?.error).toBe("provider 500"); expect(session?.entries[session.entries.length - 1]).toEqual({ kind: "report", @@ -267,6 +269,60 @@ describe("createSubAgentSessionStore", () => { // Top-level sessions carry no parent link. expect(orchestrator.parentSessionId).toBeUndefined(); }); + + test("cancel is not resumable and complete after cancel no-ops", () => { + const store = createSubAgentSessionStore({ createId: () => "s-cancel-resume" }); + store.start({ description: "stuck", agentId: "worker", brief: "loop", retained: true }); + store.markRunning("s-cancel-resume"); + store.registerFollowup("s-cancel-resume", async () => "nope"); + expect(store.cancel("s-cancel-resume", "operator kill")).toBe(true); + const session = store.get("s-cancel-resume"); + expect(session?.status).toBe("cancelled"); + expect(session?.lifecycle.state).toBe("cancelled"); + expect(session?.lifecycleStatus).toBe("interrupted"); + expect(session?.retained).toBe(false); + expect(store.resumeOne("s-cancel-resume", "more").ok).toBe(false); + store.complete("s-cancel-resume", "should not win"); + expect(store.get("s-cancel-resume")?.lifecycle.state).toBe("cancelled"); + expect(store.get("s-cancel-resume")?.report).toBeUndefined(); + }); + + test("interruptOne stays strip-running and resumable when retained", () => { + const store = createSubAgentSessionStore({ createId: () => "s-int" }); + store.start({ description: "loop", agentId: "worker", brief: "b", retained: true }); + store.markRunning("s-int"); + store.registerInterrupt("s-int", () => {}); + store.registerFollowup("s-int", async () => "next"); + expect(store.interruptOne("s-int").ok).toBe(true); + const session = store.get("s-int"); + expect(session?.lifecycle.state).toBe("interrupted"); + expect(session?.status).toBe("running"); + expect(session?.lifecycleStatus).toBe("interrupted"); + expect(store.resumeOne("s-int", "continue")).toEqual({ ok: true, status: "running" }); + }); + + test("pinned completed sessions survive maxCompleted until unpin", () => { + const store = createSubAgentSessionStore({ + maxCompleted: 1, + createId: (() => { + let n = 0; + return () => `s-${++n}`; + })(), + now: (() => { + let t = 0; + return () => ++t; + })(), + }); + const pinned = store.start({ description: "keep", agentId: "w", brief: "b" }); + store.pin(pinned.id); + store.complete(pinned.id, "keep"); + store.complete(store.start({ description: "a", agentId: "w", brief: "b" }).id, "a"); + store.complete(store.start({ description: "b", agentId: "w", brief: "b" }).id, "b"); + expect(store.get(pinned.id)).toBeDefined(); + store.unpin(pinned.id); + store.complete(store.start({ description: "c", agentId: "w", brief: "b" }).id, "c"); + expect(store.get(pinned.id)).toBeUndefined(); + }); }); describe("createTaskTool session recording", () => { From f1c851db546ce6d545d8229728db8eb37badb790 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 31 Aug 2026 16:24:33 -0700 Subject: [PATCH 2/7] Drive wait_agents from session lifecycle instead of a second store Wait JSON is a projection of stored worker lifecycle plus a per-install overlay for membership, pin, collect-freeze, and the close/send_input interrupt edges that must unblock before the session record is terminal. Spawn and resume settlement write only the session store. --- src/agent/tools.ts | 2 +- src/subagent/agent-fleet.test.ts | 89 +++++- src/subagent/agent-fleet.ts | 353 +++++++++++----------- src/subagent/lifecycle-tools.test.ts | 23 +- src/subagent/lifecycle-tools.ts | 7 +- src/subagent/lifecycle.ts | 25 ++ src/subagent/run.ts | 2 +- src/subagent/session-store.ts | 89 +++++- src/subagent/spawn-agent-worktree.test.ts | 18 +- src/subagent/task-tool.ts | 20 +- src/subagent/task-via-fleet.test.ts | 2 +- 11 files changed, 397 insertions(+), 233 deletions(-) diff --git a/src/agent/tools.ts b/src/agent/tools.ts index dcb3bc03..56a21ecf 100644 --- a/src/agent/tools.ts +++ b/src/agent/tools.ts @@ -329,7 +329,7 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise(): { function makeDeps( run: (params: RunSubAgentParams) => Promise, - opts: { cwd?: string } = {}, + opts: { cwd?: string; sessions?: ReturnType } = {}, ): AgentFleetDeps { + const sessions = opts.sessions ?? createSubAgentSessionStore(); return { permissionGate: testPermissionGate, cwd: opts.cwd ?? "/tmp", getWorkdirBase: () => "/tmp/workdir", provider, run, - sessions: createSubAgentSessionStore(), - fleetRecords: createFleetRecords(), + sessions, + fleetRecords: createFleetRecords(sessions), }; } @@ -287,12 +288,45 @@ describe("spawn_agent + wait_agents", () => { report?: string; }[]; expect(results).toHaveLength(1); - expect(results[0]!.status).toBe("done"); + expect(results[0]!.status).toBe("interrupted"); expect(results[0]!.report).toContain("## Summary"); expect(results[0]!.report).toContain("## Findings"); expect(results[0]!.report).toContain("gate.ts"); // Strip stays cancelled — salvage is for wait_agents, not a resurrection. expect(deps.sessions.get(id)?.status).toBe("cancelled"); + expect(deps.sessions.get(id)?.lifecycle.state).toBe("cancelled"); + }); + + test("catch cancel wait_agents is interrupted, not failed", async () => { + const deps = makeDeps(async (params) => { + await new Promise((resolve) => { + if (params.signal?.aborted) { + resolve(); + return; + } + params.signal?.addEventListener("abort", () => resolve(), { once: true }); + }); + const err = new Error("aborted"); + err.name = "AbortError"; + throw err; + }); + const spawn = createSpawnAgentTool(deps); + const wait = createWaitAgentsTool({ sessions: deps.sessions, fleetRecords: deps.fleetRecords }); + + const spawned = await callTool(spawn, { + description: "catch cancel", + prompt: "probe", + intent: "explore", + }); + const id = spawned.agent_id as string; + expect(deps.sessions.cancel(id)).toBe(true); + + const waited = await callTool(wait, { targets: [id], timeout_ms: 5000 }); + expect(waited.timed_out).toBe(false); + const results = waited.results as { status: string; error?: string; report?: string }[]; + expect(results[0]!.status).toBe("interrupted"); + expect(results[0]!.error).toBeUndefined(); + expect(deps.sessions.get(id)?.status).toBe("cancelled"); }); }); @@ -733,10 +767,9 @@ describe("interrupt_agent unblocks wait_agents", () => { }); const id = spawned.agent_id as string; - // Soft-interrupt via the session store only — leave fleetRecords running - // so wait_agents takes the soft-path fallback (not the terminal-record branch). + // interruptOne is wait-terminal via session interrupted; collect freezes it. expect(deps.sessions.interruptOne(id).ok).toBe(true); - expect(deps.fleetRecords.peek(id)?.status).toBe("running"); + expect(deps.fleetRecords.peek(id)?.status).toBe("interrupted"); const waited = await callTool(wait, { targets: [id], timeout_ms: 5000 }); expect(waited.timed_out).toBe(false); @@ -805,7 +838,7 @@ describe("interrupt_agent unblocks wait_agents", () => { test("soft-interrupt wait collects so a later followup cannot resurrect done", async () => { const sessions = createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(); + const fleetRecords = createFleetRecords(sessions); const worker = sessions.start({ id: "soft-int", description: "looping", @@ -814,8 +847,6 @@ describe("interrupt_agent unblocks wait_agents", () => { retained: true, }); sessions.markRunning(worker.id); - // Running fleet record + soft-interrupted session (lifecycle only) — - // the wait soft path must interrupt+take before returning. fleetRecords.register(worker.id); sessions.registerInterrupt(worker.id, () => {}); sessions.interruptOne(worker.id); @@ -828,9 +859,47 @@ describe("interrupt_agent unblocks wait_agents", () => { expect(fleetRecords.peek(worker.id)?.collected).toBe(true); fleetRecords.completeAfterInterrupt(worker.id, "resurrected reply"); + sessions.complete(worker.id, "resurrected reply"); expect(fleetRecords.peek(worker.id)?.status).toBe("interrupted"); expect(fleetRecords.peek(worker.id)?.collected).toBe(true); }); + + test("uncollected send_input followup complete clears overlay so wait is done", async () => { + const followupGate = deferred(); + const gate = deferred(); + const deps = makeDeps(async (params) => { + params.onAgentReady?.({ + close: async () => {}, + interrupt: () => {}, + followup: async () => followupGate.promise, + deliver: () => {}, + }); + return gate.promise; + }); + const spawn = createSpawnAgentTool(deps); + const wait = createWaitAgentsTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + const sendInput = createSendInputTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + const spawned = await callTool(spawn, { + description: "looping", + prompt: "do it", + intent: "explore", + }); + const id = spawned.agent_id as string; + await callTool(sendInput, { target: id, message: "stop that", interrupt: true }); + followupGate.resolve("followup report"); + await new Promise((resolve) => setTimeout(resolve, 20)); + const waited = await callTool(wait, { targets: [id], timeout_ms: 5000 }); + expect(waited.timed_out).toBe(false); + const results = waited.results as { status: string; report?: string }[]; + expect(results[0]!.status).toBe("done"); + expect(results[0]!.report).toBe("followup report"); + }); }); describe("close_agent unblocks wait_agents", () => { diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index 26d18aed..6e73b13d 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -8,26 +8,22 @@ * own workers (wait_agents), instead of one task() call per worker * serializing the wait. * - * Running state and the mailbox (`subscribe`) are the existing - * SubAgentSessionStore's — wait_agents' blocking is driven by that - * `subscribe` raced against a timeout timer, never polling. But the store's - * finished-session retention is a TUI display cap (`maxCompleted`, default - * 20): `complete()`/`fail()` evict the oldest finished session — report and - * all — once more than that many have finished. task() never hit this - * because it awaits its own single result before the tool call returns; here - * a caller can spawn far more workers than the cap in one turn and only + * Running state is the session store's `WorkerLifecycle`. wait_agents blocks + * on that store's `subscribe` raced against a timeout timer, never polling. + * Wait JSON is a projection of stored lifecycle plus a per-install overlay + * (`fleetRecords`): membership, pin, collected, and an optional wait-status + * override. Spawn/resume settlement writes only the session store. + * + * The store's finished-session retention is a TUI display cap (`maxCompleted`, + * default 20): `complete()`/`fail()` evict the oldest finished session — + * report and all — once more than that many have finished. task() never hit + * this because it awaits its own single result before the tool call returns; + * here a caller can spawn far more workers than the cap in one turn and only * `wait_agents` them later, so an evicted report would otherwise vanish - * silently. `fleetRecords` below is a small, deliberately-separate map - * (agent id -> terminal status/report/error), kept alive across the store's - * own eviction and cleared only when `wait_agents` delivers a result to a - * caller — it exists precisely because the store's cap cannot be trusted for - * this use. Its heavy payloads (report/error text) are capped at - * `MAX_FLEET_RECORDS`: past that, the oldest already-collected entry is - * compacted to a tombstone (status only, plus a pointer at - * `read_agent_trace` for the detail), falling back to the oldest - * uncollected one only once every collected entry is gone — a caller who - * never called wait_agents still gets a terminal status, never a bare - * "unknown". + * silently. Overlay `register` pins the session (honored by pruneCompleted + * and pruneRetained) until collect unpins. Heavy payloads are still capped at + * `MAX_FLEET_RECORDS`: past that, the oldest never-collected pin is compacted + * to a tombstone (status only, plus a pointer at `read_agent_trace`). * * Argument shape intentionally mirrors `task()`'s (description/prompt/ * context/goals/intent/success_criteria/do_not/report_focus) so a @@ -64,6 +60,7 @@ import { resolveEffortForRole } from "../provider/reasoning-effort.js"; import { isCodexProviderName } from "../config/codex-providers.js"; import { buildDispatchBrief, type TaskIntent } from "./report.js"; import { DEFAULT_CANCEL_REASON, type SubAgentSessionStore } from "./session-store.js"; +import { projectWaitStatus, type WaitJSONStatus } from "./lifecycle.js"; import type { NestedDispatchDeps, RunSubAgentParams, @@ -85,9 +82,9 @@ import { isSubAgentCancelError } from "./dispose.js"; const log = getLogger([LOG_NAMESPACE_ROOT, "subagent", "agent-fleet"]); -/** Terminal (or running) record for one spawned agent, keyed by agent id. */ +/** Wait JSON projection for one spawned agent, keyed by agent id. */ interface FleetRecord { - status: "running" | "done" | "failed" | "interrupted"; + status: WaitJSONStatus; report?: string; error?: string; /** Set once a wait_agents caller has been handed this result. */ @@ -98,21 +95,38 @@ interface FleetRecord { hint?: string; } +/** Per-install overlay: membership, pin, collected, optional wait override. */ +interface FleetOverlay { + collected?: boolean; + pinHeld?: boolean; + /** send_input interrupt:true / close_agent — wait interrupted while session may still be running. */ + forceInterrupted?: boolean; + /** Frozen wait status after collect. Later session completed must not resurrect this mailbox. */ + frozenStatus?: WaitJSONStatus; + tombstoned?: boolean; + hint?: string; +} + const RECOVERY_HINT = "Report evicted to bound fleet memory; recover full detail via read_agent_trace(agent_id)."; -/** Payload cap: terminal records still holding a report/error. */ +/** Payload cap: uncollected pinned terminal records still holding a report. */ export const MAX_FLEET_RECORDS = 200; /** - * Terminal-result store, cleared once a result is delivered to a - * wait_agents caller. See the module doc comment for why the session - * store's own retention cannot be reused here, and for the tombstone - * eviction policy once more than `MAX_FLEET_RECORDS` payloads are held. + * Per-install wait overlay. Session lifecycle is the source of wait status + * unless this overlay forces interrupted or has frozen a collected result. + * See the module doc for pin/tombstone policy. */ class FleetRecords { - private readonly records = new Map(); + private readonly records = new Map(); private readonly listeners = new Set<() => void>(); + private readonly sessions: SubAgentSessionStore; + + constructor(sessions: SubAgentSessionStore) { + this.sessions = sessions; + sessions?.subscribe(() => this.enforceCap()); + } subscribe(listener: () => void): () => void { this.listeners.add(listener); @@ -126,64 +140,54 @@ class FleetRecords { } register(id: string): void { - this.records.set(id, { status: "running" }); - } - - resolve(id: string, report: string): void { const existing = this.records.get(id); - if (existing !== undefined && existing.status !== "running") return; - this.records.set(id, { status: "done", report }); + const alreadyPinned = + existing !== undefined && existing.collected !== true && existing.pinHeld === true; + this.records.set(id, { pinHeld: true }); + if (!alreadyPinned) this.sessions?.pin(id); this.enforceCap(); - this.notify(); } - reject(id: string, error: string): void { - const existing = this.records.get(id); - if (existing !== undefined && existing.status !== "running") return; - this.records.set(id, { status: "failed", error }); - this.enforceCap(); - this.notify(); - } + /** + * Leftover dual-write. Settlement writes the session store; commit 3 deletes this. + */ + resolve(_id: string, _report: string): void {} + + /** + * Leftover dual-write. Settlement writes the session store; commit 3 deletes this. + */ + reject(_id: string, _error: string): void {} /** - * Marks a still-running record interrupted so wait_agents unblocks. - * No-op on an already-terminal id that is not interrupted — a late - * interrupt after complete/fail is meaningless. A late salvage report may - * still attach to an interrupted record that has none yet (including after - * an early collect), but never overwrites an existing report. + * Overlay wait-status override so wait unblocks while the session may still + * be running (send_input interrupt:true followup, close_agent teardown). + * No-op on an already-collected mailbox — frozen status stays interrupted. */ - interrupt(id: string, report?: string): void { + interrupt(id: string, _report?: string): void { const existing = this.records.get(id); if (existing === undefined) return; - if ( - existing.status === "interrupted" && - report !== undefined && - existing.report === undefined - ) { - existing.report = report; + if (existing.collected === true) { + this.sessions?.wake(); this.notify(); return; } - if (existing.status !== "running") return; - this.records.set(id, { - status: "interrupted", - ...(report !== undefined ? { report } : {}), - }); + existing.forceInterrupted = true; + this.sessions?.wake(); this.enforceCap(); this.notify(); } /** - * send_input interrupt:true queued a followup that has now finished. - * Upgrade an uncollected interrupted record to done. No-op if wait_agents - * already collected the interrupt, so a later reply cannot resurrect it. + * send_input interrupt:true followup finished. Clear an uncollected + * interrupted overlay so wait projects session completed → done. No-op if + * wait_agents already collected the interrupt. */ - completeAfterInterrupt(id: string, report: string): void { + completeAfterInterrupt(id: string, _report?: string): void { const existing = this.records.get(id); if (existing === undefined || existing.collected === true) return; - if (existing.status !== "interrupted") return; - this.records.set(id, { status: "done", report }); - this.enforceCap(); + if (existing.forceInterrupted !== true) return; + delete existing.forceInterrupted; + this.sessions?.wake(); this.notify(); } @@ -200,57 +204,101 @@ class FleetRecords { /** Read without consuming — used for the terminal-yet check. */ peek(id: string): FleetRecord | undefined { - return this.records.get(id); + if (!this.records.has(id)) return undefined; + return this.snapshot(id); } /** - * Read and, if terminal, mark collected. The entry is kept (not deleted) - * so a later query still resolves to a real status instead of "unknown" — - * it just becomes the preferred eviction target once the payload cap is - * hit. + * Read and, if terminal, freeze wait status and mark collected. Does not + * snapshot an empty report so late interrupt salvage can still attach. + * Collect unpins. */ take(id: string): FleetRecord | undefined { - const record = this.records.get(id); - if (record !== undefined && record.status !== "running") { - record.collected = true; + const overlay = this.records.get(id); + if (overlay === undefined) return undefined; + const snap = this.snapshot(id); + if (snap.status !== "running" && overlay.collected !== true) { + overlay.frozenStatus = snap.status; + overlay.collected = true; + if (overlay.pinHeld === true) { + overlay.pinHeld = false; + this.sessions?.unpin(id); + } + } + return this.snapshot(id); + } + + private sessionWaitStatus(id: string): WaitJSONStatus | undefined { + const session = this.sessions?.get(id); + if (session === undefined) return undefined; + return projectWaitStatus(session.lifecycle, this.sessions?.isRunInFlight(id) === true); + } + + private projectedStatus(id: string, overlay: FleetOverlay): WaitJSONStatus { + if (overlay.frozenStatus !== undefined) return overlay.frozenStatus; + if (overlay.forceInterrupted === true) return "interrupted"; + return this.sessionWaitStatus(id) ?? "running"; + } + + snapshot(id: string): FleetRecord { + const overlay = this.records.get(id); + if (overlay === undefined) { + return { status: "running" }; } - return record; + const status = this.projectedStatus(id, overlay); + const session = this.sessions?.get(id); + const sessionWait = this.sessionWaitStatus(id); + const payload = + overlay.tombstoned !== true && session !== undefined && sessionWait === status + ? session + : undefined; + return { + status, + ...(overlay.collected === true ? { collected: true } : {}), + ...(overlay.tombstoned === true ? { tombstoned: true } : {}), + ...(overlay.hint !== undefined ? { hint: overlay.hint } : {}), + ...(payload?.report !== undefined ? { report: payload.report } : {}), + ...(payload?.error !== undefined && status === "failed" ? { error: payload.error } : {}), + }; } - private hasPayload(record: FleetRecord): boolean { - return record.status !== "running" && !record.tombstoned; + private isPayload(id: string, overlay: FleetOverlay): boolean { + if (overlay.tombstoned === true || overlay.collected === true) return false; + if (overlay.pinHeld !== true) return false; + return this.projectedStatus(id, overlay) !== "running"; } /** - * Compacts the oldest already-collected payload to a tombstone first — - * its caller already has the detail — and only reaches into uncollected - * payloads once no collected one remains. + * Compacts the oldest never-collected pin to a tombstone once more than + * `MAX_FLEET_RECORDS` terminal payloads are held. */ private enforceCap(): void { - let payloadCount = 0; - for (const record of this.records.values()) { - if (this.hasPayload(record)) payloadCount++; + const payloads: string[] = []; + for (const [id, overlay] of this.records) { + if (this.isPayload(id, overlay)) payloads.push(id); } - while (payloadCount > MAX_FLEET_RECORDS) { - const victim = - [...this.records.values()].find((r) => this.hasPayload(r) && r.collected === true) ?? - [...this.records.values()].find((r) => this.hasPayload(r)); + while (payloads.length > MAX_FLEET_RECORDS) { + const victimId = payloads.shift(); + if (victimId === undefined) break; + const victim = this.records.get(victimId); if (victim === undefined) break; - delete victim.report; - delete victim.error; + victim.frozenStatus = this.projectedStatus(victimId, victim); victim.tombstoned = true; victim.hint = RECOVERY_HINT; - payloadCount--; + if (victim.pinHeld === true) { + victim.pinHeld = false; + this.sessions?.unpin(victimId); + } } } } -// One registry per orchestrator install (shared by its spawn_agent and +// One overlay per orchestrator install (shared by its spawn_agent and // wait_agents tool instances), not a module singleton — created in // createSpawnAgentTool and threaded to createWaitAgentsTool by the caller. export type FleetRecordsHandle = FleetRecords; -export function createFleetRecords(): FleetRecordsHandle { - return new FleetRecords(); +export function createFleetRecords(sessions: SubAgentSessionStore): FleetRecordsHandle { + return new FleetRecords(sessions); } const SpawnAgentArgs = type({ @@ -603,7 +651,6 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { err instanceof WorktreeError ? err.message : `sub-agent worktree setup failed: ${err instanceof Error ? err.message : String(err)}`; - deps.fleetRecords.reject(session.id, message); deps.sessions.fail(session.id, message); finalizeEnd(true); return fleetResult(call.id, `Error: ${message}`); @@ -723,39 +770,35 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { }; // Fire and forget: this handler must return before the worker finishes. - // fleetRecords is the durable source of truth wait_agents reads from - // (see the module doc for its own cap/eviction policy); - // deps.sessions.complete/fail is still called for the TUI's benefit, - // but only after fleetRecords already has the result, so the - // synchronous subscribe notification always sees the up-to-date - // record. + // Wait JSON projects stored WorkerLifecycle plus the per-install overlay. deps .run(params) .then((result) => { - // interrupt_agent / send_input already flipped this session - // synchronously (session-store.interruptOne / sendInputOne) — do not - // let the settling promise's normal bookkeeping overwrite that with - // a "completed" status, and do not re-stamp the interrupt either: a - // follow-up turn may already be live on this lane. Still terminalize - // fleetRecords so a waiter that never saw interrupt_agent (or raced - // it) cannot hang. if (result.interrupted === true) { keepWorktreeAlive = true; runInterrupted = true; - deps.fleetRecords.interrupt(session.id, result.report); + const now = deps.sessions.get(session.id); + const overlay = deps.fleetRecords.peek(session.id); + // send_input interrupt:true already started a followup (session + // running + overlay interrupted). Do not stamp that turn interrupted + // or clear its in-flight bit. + const followupLive = + now?.lifecycle.state === "running" && overlay?.status === "interrupted"; + if (!followupLive) { + deps.sessions.attachReport(session.id, result.report); + } + return; + } + const alreadyCancelled = deps.sessions.get(session.id)?.status === "cancelled"; + if (alreadyCancelled) { + deps.sessions.attachReport(session.id, result.report); return; } - // Operator cancel may race after run resolves (childCtl aborted). - // Keep strip status cancelled when sessions.cancel already flipped - // it, but never discard a returned body (including salvage) — - // wait_agents reads fleetRecords, not the strip. - deps.fleetRecords.resolve(session.id, result.report); // result.agentRetained is only true on run.ts's clean-completion // path when persist actually skipped teardown — a deadline/cancel // salvage resolves through the same promise but always disposed // its agent first, so the store must not treat it as resumable // just because retained:true was requested at spawn. - // complete() no-ops when status is already cancelled. const agentRetained = result.agentRetained === true; if (agentRetained) keepWorktreeAlive = true; deps.sessions.complete(session.id, result.report, { @@ -764,24 +807,18 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { }); }) .catch((err) => { - // Always terminalize fleetRecords — including pre-progress cancel that - // rethrows with no salvage — so wait_agents does not hang. Prefer - // cancel semantics over fail when the strip already cancelled or the - // throw is an AbortError (legacy task() parent contract). const alreadyCancelled = deps.sessions.get(session.id)?.status === "cancelled"; if (alreadyCancelled || isSubAgentCancelError(err, childCtl.signal)) { if (!alreadyCancelled) { deps.sessions.cancel(session.id, DEFAULT_CANCEL_REASON); } - const message = err instanceof Error ? err.message : String(err); - deps.fleetRecords.reject(session.id, message); + deps.sessions.settleRun(session.id); return; } // Auth failures keep the actionable Re-authenticate wording that // task()'s fused path surfaces via formatSubAgentTaskAuthFailureMessage. const authMessage = formatSubAgentTaskAuthFailureMessage(description, err); const failReason = authMessage ?? (err instanceof Error ? err.message : String(err)); - deps.fleetRecords.reject(session.id, failReason); deps.sessions.fail(session.id, failReason); }) .finally(() => { @@ -799,35 +836,17 @@ interface WaitAgentsDeps { fleetRecords: FleetRecordsHandle; } -function isSoftInterrupted( - session: ReturnType, -): session is NonNullable> { - // interrupt_agent keeps strip status "running" so resume_agent can reuse - // the session. cancel() also sets lifecycleStatus "interrupted" but flips - // status to "cancelled" — that path still owes wait_agents a salvage - // report via fleetRecords, so it is not wait-terminal on its own. - return ( - session !== undefined && - session.status === "running" && - (session.lifecycleStatus === "interrupted" || session.lifecycleStatus === "shutdown") - ); -} - -function isWaitTerminal( - id: string, - sessions: SubAgentSessionStore, - fleetRecords: FleetRecordsHandle, -): boolean { +function isWaitTerminal(id: string, fleetRecords: FleetRecordsHandle): boolean { const record = fleetRecords.peek(id); - if (record !== undefined && record.status !== "running") return true; - return isSoftInterrupted(sessions.get(id)); + return record !== undefined && record.status !== "running"; } /** * Blocks until `mode` is satisfied for `targets`, or `timeoutMs` / abort * elapses. Driven by the session store's mailbox (`subscribe`) raced against * a timer and the parent tool signal; never polls. Timeout and abort have no - * side effects: workers keep running and remain waitable. + * side effects: workers keep running and remain waitable. Overlay writers + * wake this wait via `sessions.wake()`. */ async function waitForTerminal( sessions: SubAgentSessionStore, @@ -839,8 +858,8 @@ async function waitForTerminal( ): Promise { const ready = (): boolean => mode === "all" - ? targets.every((id) => isWaitTerminal(id, sessions, fleetRecords)) - : targets.some((id) => isWaitTerminal(id, sessions, fleetRecords)); + ? targets.every((id) => isWaitTerminal(id, fleetRecords)) + : targets.some((id) => isWaitTerminal(id, fleetRecords)); if (signal?.aborted) return true; if (ready()) return false; @@ -851,7 +870,6 @@ async function waitForTerminal( settled = true; clearTimeout(timer); unsubscribeSessions(); - unsubscribeFleet(); signal?.removeEventListener("abort", onAbort); resolve(timedOut); }; @@ -861,7 +879,6 @@ async function waitForTerminal( }; const timer = setTimeout(() => finish(true), timeoutMs); const unsubscribeSessions = sessions.subscribe(onChange); - const unsubscribeFleet = fleetRecords.subscribe(onChange); signal?.addEventListener("abort", onAbort, { once: true }); if (signal?.aborted) finish(true); }); @@ -897,42 +914,24 @@ export function createWaitAgentsTool(deps: WaitAgentsDeps): AgentTool { signal, ); - // Terminal fleet records are marked collected once delivered here; a - // running record is only peeked, so it stays waitable. Session - // lifecycle is a fallback for interrupt/close that raced the record. + // Terminal overlay/session projections are marked collected once + // delivered here; a running record is only peeked, so it stays waitable. const results = targets.map((id) => { const record = deps.fleetRecords.peek(id); - if (record !== undefined && record.status !== "running") { - const taken = deps.fleetRecords.take(id) ?? record; - return { - agent_id: id, - status: taken.status, - ...(taken.report !== undefined ? { report: taken.report } : {}), - ...(taken.error !== undefined ? { error: taken.error } : {}), - ...(taken.hint !== undefined ? { hint: taken.hint } : {}), - }; - } - const session = deps.sessions.get(id); - if (isSoftInterrupted(session)) { - // Match the mailbox to what we report (include salvage report when - // present), then collect so a later completeAfterInterrupt cannot - // resurrect this wait as "done". - deps.fleetRecords.interrupt(id, session.report); - const taken = deps.fleetRecords.take(id); - return { - agent_id: id, - status: "interrupted" as const, - ...(taken?.report !== undefined - ? { report: taken.report } - : session.report !== undefined - ? { report: session.report } - : {}), - }; - } if (record === undefined) { return { agent_id: id, status: "unknown" as const }; } - return { agent_id: id, status: "running" as const }; + if (record.status === "running") { + return { agent_id: id, status: "running" as const }; + } + const taken = deps.fleetRecords.take(id) ?? record; + return { + agent_id: id, + status: taken.status, + ...(taken.report !== undefined ? { report: taken.report } : {}), + ...(taken.error !== undefined ? { error: taken.error } : {}), + ...(taken.hint !== undefined ? { hint: taken.hint } : {}), + }; }); return fleetResult(call.id, JSON.stringify({ results, timed_out: timedOut })); diff --git a/src/subagent/lifecycle-tools.test.ts b/src/subagent/lifecycle-tools.test.ts index 8f38e403..97898852 100644 --- a/src/subagent/lifecycle-tools.test.ts +++ b/src/subagent/lifecycle-tools.test.ts @@ -52,7 +52,10 @@ describe("close_agent", () => { }); } - const closeAgent = createCloseAgentTool({ sessions, fleetRecords: createFleetRecords() }); + const closeAgent = createCloseAgentTool({ + sessions, + fleetRecords: createFleetRecords(sessions), + }); const result = await callTool(closeAgent, { target: parent.id }); expect(result.status).toBe("shutdown"); @@ -90,7 +93,7 @@ describe("close_agent", () => { describe("resume_agent", () => { test("starts the next turn on a completed retained session and returns immediately", async () => { const sessions = createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(); + const fleetRecords = createFleetRecords(sessions); const retained = sessions.start({ description: "d", agentId: "a", brief: "b", retained: true }); const history: string[] = ["first task"]; let finish: (reply: string) => void = () => {}; @@ -147,7 +150,7 @@ describe("resume_agent", () => { test("resumes an interrupted retained session without calling close()", async () => { const sessions = createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(); + const fleetRecords = createFleetRecords(sessions); const worker = sessions.start({ description: "worker", agentId: "a", @@ -200,7 +203,7 @@ describe("resume_agent", () => { test("rejects a closed session and a concurrent resume of a running turn", async () => { const sessions = createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(); + const fleetRecords = createFleetRecords(sessions); const closed = sessions.start({ description: "closed", agentId: "a", @@ -255,7 +258,7 @@ describe("resume_agent", () => { test("wait_agents collects the resumed turn after resume_agent returns", async () => { const sessions = createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(); + const fleetRecords = createFleetRecords(sessions); const worker = sessions.start({ description: "worker", agentId: "a", @@ -306,7 +309,7 @@ describe("interrupt_agent", () => { const interruptAgent = createInterruptAgentTool({ sessions, - fleetRecords: createFleetRecords(), + fleetRecords: createFleetRecords(sessions), }); if (interruptAgent.kind !== "full") throw new Error("expected full tool"); @@ -403,7 +406,7 @@ describe("send_input", () => { test("rejects completed, interrupted, and closed sessions — steering is in-flight only", async () => { const sessions = createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(); + const fleetRecords = createFleetRecords(sessions); const sendInput = createSendInputTool({ sessions, fleetRecords }); if (sendInput.kind !== "full") throw new Error("expected full tool"); @@ -567,7 +570,7 @@ describe("nested lifecycle authority", () => { } const interrupt = createInterruptAgentTool({ sessions, - fleetRecords: createFleetRecords(), + fleetRecords: createFleetRecords(sessions), authority: nestAuthority(sessions, nested.id), }); expect((await callTool(interrupt, { target: child.id })).status).toBe("interrupted"); @@ -593,7 +596,7 @@ describe("nested lifecycle authority", () => { for (const s of [child, sibling]) sessions.registerClose(s.id, async () => {}); const close = createCloseAgentTool({ sessions, - fleetRecords: createFleetRecords(), + fleetRecords: createFleetRecords(sessions), authority: nestAuthority(sessions, nested.id), }); expect((await callTool(close, { target: child.id })).status).toBe("shutdown"); @@ -630,7 +633,7 @@ describe("nested lifecycle authority", () => { } const resume = createResumeAgentTool({ sessions, - fleetRecords: createFleetRecords(), + fleetRecords: createFleetRecords(sessions), authority: nestAuthority(sessions, nested.id), }); expect((await callTool(resume, { target: child.id, message: "more" })).status).toBe("running"); diff --git a/src/subagent/lifecycle-tools.ts b/src/subagent/lifecycle-tools.ts index d50130b3..f990f91e 100644 --- a/src/subagent/lifecycle-tools.ts +++ b/src/subagent/lifecycle-tools.ts @@ -231,10 +231,10 @@ export function createResumeAgentTool(deps: ResumeAgentToolDeps): AgentTool { deps.fleetRecords.register(target); }, onReply: (reply) => { - deps.fleetRecords.resolve(target, reply); + deps.sessions.complete(target, reply); }, onFail: (err) => { - deps.fleetRecords.reject(target, err instanceof Error ? err.message : String(err)); + deps.sessions.fail(target, err instanceof Error ? err.message : String(err)); }, }); if (!outcome.ok) { @@ -293,9 +293,6 @@ export function createInterruptAgentTool(deps: InterruptAgentToolDeps): AgentToo `Error: cannot interrupt "${target}" (status: ${outcome.status}).`, ); } - // Wait mailbox is separate from the TUI strip — flip it here so - // wait_agents does not stay blocked on a still-"running" record. - deps.fleetRecords.interrupt(target); return lifecycleResult( call.id, JSON.stringify({ agent_id: target, status: "interrupted" satisfies AgentLifecycleStatus }), diff --git a/src/subagent/lifecycle.ts b/src/subagent/lifecycle.ts index b7c7daf4..0ea29786 100644 --- a/src/subagent/lifecycle.ts +++ b/src/subagent/lifecycle.ts @@ -75,3 +75,28 @@ export function isResumableLifecycle( retained === true && (lifecycle.state === "completed" || lifecycle.state === "interrupted") ); } + +export type WaitJSONStatus = "running" | "done" | "failed" | "interrupted"; + +/** + * Wait JSON projection of stored lifecycle. Operator cancel (`cancelled`) is + * wait-running while a run/followup is still in flight so the first collect + * can still attach salvage. `interrupted` and `shutdown` are immediately + * terminal. Never leaks `cancelled` into wait JSON. + */ +export function projectWaitStatus(lifecycle: WorkerLifecycle, inFlight: boolean): WaitJSONStatus { + if (lifecycle.state === "cancelled" && inFlight) return "running"; + switch (lifecycle.state) { + case "completed": + return "done"; + case "failed": + return "failed"; + case "interrupted": + case "cancelled": + case "shutdown": + return "interrupted"; + case "pending_init": + case "running": + return "running"; + } +} diff --git a/src/subagent/run.ts b/src/subagent/run.ts index f6a496ff..28bf2107 100644 --- a/src/subagent/run.ts +++ b/src/subagent/run.ts @@ -599,7 +599,7 @@ async function runSubAgentInner( } const nd = params.nestedDispatch; const fleetSessions = nd.sessions ?? createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(); + const fleetRecords = createFleetRecords(fleetSessions); tools = [ ...tools, createTaskTool({ diff --git a/src/subagent/session-store.ts b/src/subagent/session-store.ts index aa0235a0..a5510782 100644 --- a/src/subagent/session-store.ts +++ b/src/subagent/session-store.ts @@ -234,6 +234,23 @@ export interface SubAgentSessionStore { */ pin(id: string): void; unpin(id: string): void; + /** + * Attach a salvage report to cancelled/interrupted/shutdown without changing + * `state`. Never overwrites an existing report. If the session is still + * pending_init/running (interrupt result with no prior interrupt_agent), + * flip to interrupted rather than completed. Clears the in-flight-run bit + * and notifies waiters. + */ + attachReport(id: string, report: string): void; + /** True while a run or followup has not settled. */ + isRunInFlight(id: string): boolean; + /** + * Catch-path settle: clear the in-flight bit without changing lifecycle so + * operator cancel becomes wait-terminal when there is no salvage body. + */ + settleRun(id: string): void; + /** Wake subscribers without mutating a session (mailbox overlay writers). */ + wake(): void; subscribe(listener: () => void): () => void; clear(): void; } @@ -387,6 +404,8 @@ export function createSubAgentSessionStore( const sessions = new Map(); // Pin refcount: wait mailboxes hold a pin until they collect the result. const pinCounts = new Map(); + // Live run/followup: operator cancel is wait-terminal only after this clears. + const runInFlight = new Set(); // Live abort hooks keyed by session id. Cleared on terminal transition. const cancelHandles = new Map void>(); // CL-6943: bounded close functions keyed by session id, for close_agent. @@ -570,7 +589,7 @@ export function createSubAgentSessionStore( // still report an actionable status afterward instead of "not_found". const pruneRetained = (): void => { const openRetained = [...sessions.values()] - .filter(isOpenRetained) + .filter((s) => isOpenRetained(s) && !isPinned(s.id)) .sort((a, b) => a.lastActivityAt - b.lastActivityAt); const excess = openRetained.length - maxRetained; if (excess <= 0) return; @@ -633,6 +652,7 @@ export function createSubAgentSessionStore( // resume_agent can retry. interrupt_agent's stamp on this turn wins over // that restore — do not rewrite interrupted back to completed. const beginFollowupTurn = (id: string): void => { + runInFlight.add(id); mutate(id, (s) => { s.lifecycle = { state: "running" }; delete s.finishedAt; @@ -667,12 +687,16 @@ export function createSubAgentSessionStore( void followup(message) .then((reply) => { const still = sessions.get(id); - if (still === undefined) return; + if (still === undefined) { + runInFlight.delete(id); + return; + } if ( still.lifecycle.state === "shutdown" || still.lifecycle.state === "cancelled" || still.lifecycle.state === "failed" ) { + runInFlight.delete(id); return; } mutate(id, (s) => { @@ -681,12 +705,17 @@ export function createSubAgentSessionStore( s.report = reply; pushEntry(s, { kind: "report", content: capText(reply, maxEntryChars) }); }); + runInFlight.delete(id); opts?.onReply?.(reply); pruneRetained(); }) .catch((err: unknown) => { - endFollowupTurn(id, failLifecycle); - opts?.onFail?.(err); + runInFlight.delete(id); + if (opts?.onFail !== undefined) { + opts.onFail(err); + } else { + endFollowupTurn(id, failLifecycle); + } log.error("followup turn failed for {id}: {error}", { id, error: err instanceof Error ? err.message : String(err), @@ -723,6 +752,7 @@ export function createSubAgentSessionStore( followupHandles.delete(id); deliverHandles.delete(id); pinCounts.delete(id); + runInFlight.delete(id); forgetRevision(id); const session: StoredSession = { id, @@ -742,6 +772,7 @@ export function createSubAgentSessionStore( ...(input.parentSessionId !== undefined ? { parentSessionId: input.parentSessionId } : {}), }; sessions.set(id, session); + runInFlight.add(id); bumpRevision(id); notify(); return snapshotOf(session); @@ -921,6 +952,7 @@ export function createSubAgentSessionStore( // release it now rather than leaving a stale reference around. cancelHandles.delete(id); if (!agentRetained) closeHandles.delete(id); + runInFlight.delete(id); pruneCompleted(); pruneRetained(); }); @@ -948,6 +980,7 @@ export function createSubAgentSessionStore( }); cancelHandles.delete(id); closeHandles.delete(id); + runInFlight.delete(id); pruneCompleted(); }); }, @@ -1036,6 +1069,7 @@ export function createSubAgentSessionStore( interruptHandles.delete(id); followupHandles.delete(id); deliverHandles.delete(id); + runInFlight.delete(id); pruneCompleted(); return "shutdown"; }, @@ -1184,8 +1218,50 @@ export function createSubAgentSessionStore( unpin(id: string): void { const next = (pinCounts.get(id) ?? 0) - 1; - if (next <= 0) pinCounts.delete(id); - else pinCounts.set(id, next); + if (next <= 0) { + pinCounts.delete(id); + pruneCompleted(); + pruneRetained(); + } else pinCounts.set(id, next); + }, + + attachReport(id: string, report: string): void { + mutate(id, (session) => { + const state = session.lifecycle.state; + if (state === "completed" || state === "failed") { + runInFlight.delete(id); + return; + } + if (state === "pending_init" || state === "running") { + session.lifecycle = { state: "interrupted", report }; + session.report = report; + session.finishedAt = session.finishedAt ?? now(); + pushEntry(session, { kind: "report", content: capText(report, maxEntryChars) }); + } else if ( + (state === "cancelled" || state === "interrupted" || state === "shutdown") && + session.report === undefined + ) { + session.report = report; + session.lifecycle = { ...session.lifecycle, report }; + pushEntry(session, { kind: "report", content: capText(report, maxEntryChars) }); + } + runInFlight.delete(id); + pruneCompleted(); + pruneRetained(); + }); + }, + + isRunInFlight(id: string): boolean { + return runInFlight.has(id); + }, + + settleRun(id: string): void { + if (!runInFlight.delete(id)) return; + notify(); + }, + + wake(): void { + notify(); }, subscribe(listener: () => void): () => void { @@ -1207,6 +1283,7 @@ export function createSubAgentSessionStore( deliverHandles.clear(); sessions.clear(); pinCounts.clear(); + runInFlight.clear(); revisions.clear(); snapshotCache.clear(); evicted.clear(); diff --git a/src/subagent/spawn-agent-worktree.test.ts b/src/subagent/spawn-agent-worktree.test.ts index 2df03297..52c703b9 100644 --- a/src/subagent/spawn-agent-worktree.test.ts +++ b/src/subagent/spawn-agent-worktree.test.ts @@ -93,6 +93,7 @@ describe("spawn_agent worktree isolation", () => { tempDirs.push(workdirBase); let captured: RunSubAgentParams | undefined; + const sessions = createSubAgentSessionStore(); const tool = createSpawnAgentTool({ permissionGate: testPermissionGate, cwd: repo, @@ -103,8 +104,8 @@ describe("spawn_agent worktree isolation", () => { captured = params; return { report: "done" }; }, - sessions: createSubAgentSessionStore(), - fleetRecords: createFleetRecords(), + sessions, + fleetRecords: createFleetRecords(sessions), }); if (tool.kind !== "full") throw new Error("expected full tool"); const result = await tool.handler( @@ -143,7 +144,7 @@ describe("spawn_agent worktree isolation", () => { return { report: "no" }; }, sessions, - fleetRecords: createFleetRecords(), + fleetRecords: createFleetRecords(sessions), }); if (tool.kind !== "full") throw new Error("expected full tool"); const result = await tool.handler( @@ -208,7 +209,7 @@ describe("spawn_agent worktree isolation", () => { throw error; }, sessions, - fleetRecords: createFleetRecords(), + fleetRecords: createFleetRecords(sessions), }); if (tool.kind !== "full") throw new Error("expected full tool"); @@ -259,7 +260,7 @@ describe("spawn_agent worktree isolation", () => { return settle.promise; }, sessions, - fleetRecords: createFleetRecords(), + fleetRecords: createFleetRecords(sessions), }); if (tool.kind !== "full") throw new Error("expected full tool"); const spawned = await tool.handler( @@ -332,7 +333,7 @@ describe("spawn_agent worktree isolation", () => { return result; }, sessions, - fleetRecords: createFleetRecords(), + fleetRecords: createFleetRecords(sessions), }); if (tool.kind !== "full") throw new Error("expected full tool"); const spawned = await tool.handler( @@ -379,6 +380,7 @@ describe("spawn_agent worktree isolation", () => { tempDirs.push(workdirBase); let workerCwd: string | undefined; + const sessions = createSubAgentSessionStore(); const tool = createSpawnAgentTool({ permissionGate: testPermissionGate, cwd: repo, @@ -390,8 +392,8 @@ describe("spawn_agent worktree isolation", () => { // Salvage / non-persist path: no agentRetained flag. return { report: "## Summary\nSalvaged." }; }, - sessions: createSubAgentSessionStore(), - fleetRecords: createFleetRecords(), + sessions, + fleetRecords: createFleetRecords(sessions), }); if (tool.kind !== "full") throw new Error("expected full tool"); await tool.handler( diff --git a/src/subagent/task-tool.ts b/src/subagent/task-tool.ts index d64a24bb..91e457dc 100644 --- a/src/subagent/task-tool.ts +++ b/src/subagent/task-tool.ts @@ -403,26 +403,17 @@ async function runTaskViaFleet(input: { if (result === undefined) { return taskToolResult(input.callId, `Error: wait_agents returned no result for ${agentId}.`); } - // Cancel must not be misclassified as failed:abort — strip cancel / - // AbortError leaves fleetRecords as failed with "aborted" while the - // session store holds cancelled + the operator reason. A cancel that - // still resolved a salvage body (fleet status done) keeps the report, - // matching the fused task() race contract. - const session = input.sessions.get(agentId); - if (session?.status === "cancelled") { - if ( - (result.status === "done" || result.status === "interrupted") && - typeof result.report === "string" && - result.report.length > 0 - ) { + if (result.status === "interrupted") { + if (typeof result.report === "string" && result.report.length > 0) { return taskToolResult( input.callId, `Sub-agent "${input.description}" reported:\n\n${result.report}`, ); } + const session = input.sessions.get(agentId); return taskToolResult( input.callId, - cancelledSubAgentMessage(input.description, session.error), + cancelledSubAgentMessage(input.description, session?.error), ); } if (result.status === "failed") { @@ -460,7 +451,8 @@ export function createTaskTool(deps: TaskToolDeps): AgentTool { const briefLedger = createBriefDispatchLedger(); const fleetSessions = deps.sessions; const fleetRecords = - deps.fleetRecords ?? (fleetSessions !== undefined ? createFleetRecords() : undefined); + deps.fleetRecords ?? + (fleetSessions !== undefined ? createFleetRecords(fleetSessions) : undefined); // Every completed dispatch gets an outcome record — the log otherwise // carries shape and run state but never what the run actually produced. // Tagged with the dispatched child's provider/model/family so diff --git a/src/subagent/task-via-fleet.test.ts b/src/subagent/task-via-fleet.test.ts index b6903986..098daa37 100644 --- a/src/subagent/task-via-fleet.test.ts +++ b/src/subagent/task-via-fleet.test.ts @@ -20,7 +20,7 @@ const provider = { describe("task via spawn_agent + wait_agents", () => { test("a director task with a session store returns the worker report", async () => { const sessions = createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(); + const fleetRecords = createFleetRecords(sessions); const tool = createTaskTool({ permissionGate: testPermissionGate, cwd: "/tmp", From 2b64815ba63d8e0ffee388abea023a55c8399b65 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 31 Aug 2026 16:48:16 -0700 Subject: [PATCH 3/7] Rename the wait overlay to FleetMailbox Wait JSON is a projection of stored WorkerLifecycle plus a per-install mailbox. Drop leftover resolve/reject methods that no longer write a second terminal store. Keep the dependency field name fleetRecords so call-site churn stays a rename of the type and factory. --- docs/ARCHITECTURE.md | 4 +- src/agent/tools.ts | 4 +- src/subagent/agent-fleet.test.ts | 18 ++++---- src/subagent/agent-fleet.ts | 55 +++++++---------------- src/subagent/lifecycle-tools.test.ts | 23 +++++----- src/subagent/lifecycle-tools.ts | 10 ++--- src/subagent/run.ts | 4 +- src/subagent/spawn-agent-worktree.test.ts | 14 +++--- src/subagent/task-tool.ts | 10 ++--- src/subagent/task-via-fleet.test.ts | 4 +- 10 files changed, 61 insertions(+), 85 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 23f1834f..5a225c25 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -226,7 +226,7 @@ Every director package carries a required `tier: SubagentTier` field (`src/agent Enforcement is runtime code at the existing tool-mount point, not prompt wording — this is the fix for four prior mechanisms (`writePaths`, `report.requiredSections`, a `--config` comment, the thrash matcher) that were documented-as-enforced while enforcing nothing: - **Mount-time gate — live today, and fails closed.** `task-tool.ts` resolves the caller's tier at dispatch time — a closed director's `DirectorPackage.tier` — and forwards it as `RunSubAgentParams.orchestratorTier`. `runSubAgent` (`src/subagent/run.ts`) then calls `assertTierMayMountFleetVerb(tier, toolName)` (`src/subagent/authority.ts`) before installing fleet verbs, treating a **missing** `orchestratorTier` as `"leaf"` — deny, not skip. This is the case that matters most: a project-local or plugin `AgentProfile` with `orchestrator: true` is outside the closed director set and is **not** trusted with fleet verbs just because `orchestrator: true` is set — there is no profile-level opt-in today, so the mount always throws `FleetAuthorityError` for a profile-sourced orchestrator. `FLEET_VERBS` in `authority.ts` names the live verbs (`task`, `spawn_agent`, `wait_agents`, `list_agents`, `send_input`, `interrupt_agent`, `close_agent`, `resume_agent`, `read_agent_trace`, `search_agents`) so every mount site inherits the same gate. `list_agents` is the non-blocking mailbox-scoped list of this install's own `spawn_agent` workers (same scope as `wait_agents`); nested orchestrators may mount it. Fleet discovery (`search_agents`) remains Tier 1 only. -- **Subtree authority — wired for addressing verbs.** `assertCanTargetAgent(actor, targetId, nodes)` implements the "root owns its tree; a child manages only its own descendants" rule over the `{id, parentSessionId}` shape `SubAgentSessionStore` already tracks. Production call sites: `read_agent_trace`, `send_input`, `interrupt_agent`, `close_agent`, and `resume_agent`. Nested mounts pass `{actorId, tier, getNodes}` from `run.ts`; a missing `actorId` fails closed. Tier-1 primary omits authority and stays unrestricted. `spawn_agent` records `parentSessionId` on nested workers so `close_agent`'s descendant walk can see them. `wait_agents` with omitted targets waits only on that caller's own `fleetRecords`, not every running session in the shared store. `list_agents` reports that same mailbox without blocking. `interrupt_agent` / `send_input` with `interrupt:true` terminalize the wait mailbox immediately; the soft-interrupt wait path collects so a later followup cannot resurrect an already-observed interrupt. `close_agent` also terminalizes the wait mailbox before teardown. +- **Subtree authority — wired for addressing verbs.** `assertCanTargetAgent(actor, targetId, nodes)` implements the "root owns its tree; a child manages only its own descendants" rule over the `{id, parentSessionId}` shape `SubAgentSessionStore` already tracks. Production call sites: `read_agent_trace`, `send_input`, `interrupt_agent`, `close_agent`, and `resume_agent`. Nested mounts pass `{actorId, tier, getNodes}` from `run.ts`; a missing `actorId` fails closed. Tier-1 primary omits authority and stays unrestricted. `spawn_agent` records `parentSessionId` on nested workers so `close_agent`'s descendant walk can see them. `wait_agents` with omitted targets waits only on that caller's per-install wait mailbox over the shared session store, not every running session. Wait JSON is a projection of stored lifecycle plus mailbox membership/pin/collected/interrupt override — not a second terminal store. `list_agents` reports that same mailbox without blocking. `interrupt_agent` / `send_input` with `interrupt:true` terminalize the wait mailbox immediately; the soft-interrupt wait path collects so a later followup cannot resurrect an already-observed interrupt. `close_agent` also terminalizes the wait mailbox before teardown. Operator cancel (`cancel` / `cancelAll`) projects wait status `interrupted`. - `task()` remains the deprecated fused spawn+wait fallback. `spawn_agent` + `wait_agents` is the supported parallel path. The tier check still gates which packages may mount any fleet verb. #### Closed director fleet (`src/agent/directors/`) @@ -295,6 +295,8 @@ Every shipped specialist is a **director package** — a prompt-first `DirectorP **Session records** (`src/subagent/session-store.ts`): each spawn is retained as an inspectable child session (id, profile, description, brief, status, tool activity, transcript entries). Child events land only in this store — not in the parent chat transcript. Live progress still uses the light `onProgress` channel for the status bar. Completed sessions are capped (`maxCompleted`) so a long chat does not grow without bound. +**Wait mailbox** (`src/subagent/agent-fleet.ts` `FleetMailbox`): per-install overlay over that session store. Wait JSON is a projection of stored lifecycle plus mailbox membership, pin, collected, and optional interrupt override — not a second terminal store. Mailbox `register` pins an uncollected result (honored by prune); past `MAX_FLEET_RECORDS` the oldest never-collected pin is compacted to a tombstone. Operator cancel projects wait status `interrupted`. + **Observe (OpenTUI)**: `shell.ts:enterSubagentObserve` swaps the transcript for a child's stream (live while running, historical when done) without stealing the parent reactor; child events are mapped to stream rows by `src/tui/observe-map.ts`. Esc leaves observe and restores the parent transcript. Parent Esc/stop and `/clear` still call `cancelAll` so live children close (`agent.close`) instead of continuing after the parent stops. The host-injection point that resolves a live session (`onObserveRequest` → `observeSessionFromSubAgents`, `src/tui/runner-host.ts`, picking the newest running child else the most recent session of any status) is triggered by Alt+O (`shell.ts:observeActiveSubagent`) — the command palette action that used to call it is gone along with `src/tui/palette.ts` itself, but the chord replaces it rather than dropping the feature. Data-only agent plugins (`src/plugins/data-only-agent.ts`) synthesize `agentPlugin.agents[]` from `agents/*.md` or flat `*.md` in the plugin directory, with optional co-located `skills/`. `loadPluginEntry` tries JS entrypoints first, then falls back to this layout (`/plugins` add-by-path supports filesystem completion via `listPathSuggestions`). diff --git a/src/agent/tools.ts b/src/agent/tools.ts index 56a21ecf..16ab6447 100644 --- a/src/agent/tools.ts +++ b/src/agent/tools.ts @@ -42,7 +42,7 @@ import { type SubAgentSessionStore, } from "../subagent/index.js"; import { - createFleetRecords, + createFleetMailbox, createSpawnAgentTool, createWaitAgentsTool, createListAgentsTool, @@ -329,7 +329,7 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise { test("reports survive well past the session store's display cap (20) until wait_agents collects them", async () => { // DEFAULT_MAX_COMPLETED on SubAgentSessionStore is 20 finished sessions; // spawn (and complete) enough workers to blow well past it before any of - // them is collected, proving fleetRecords does not depend on the store's - // cap. + // them is collected, proving the wait mailbox pin keeps reports past the + // store's display cap. // // CL-7007: this test previously asserted (as CL-7001's fix left it) that // the store itself had already evicted and released the earliest @@ -215,8 +215,8 @@ describe("spawn_agent + wait_agents", () => { // ticket fixes (resume_agent failed with a bare // "not_found" past 20 spawned workers, blaming the caller for nothing). // Open retained sessions now have their own cap (`maxRetained`, default - // 50), so 25 of them all stay resumable; fleetRecords/wait_agents is - // still asserted below as the durable source of truth regardless. + // 50), so 25 of them all stay resumable; mailbox pin + wait_agents is + // still asserted below as the collect path regardless. const COUNT = 25; const deps = makeDeps(async () => ({ report: "irrelevant", agentRetained: true })); const spawn = createSpawnAgentTool(deps); @@ -250,7 +250,7 @@ describe("spawn_agent + wait_agents", () => { }); // CL-6915: operator cancel aborts the child signal, but run() still returns a - // salvage body (partial findings). Dropping that body left fleetRecords + // salvage body (partial findings). Dropping that body left the wait mailbox // "running" forever so wait_agents never saw the salvage. test("cancelled spawn_agent still resolves wait_agents with salvage findings", async () => { const deps = makeDeps(async (params) => { @@ -356,7 +356,7 @@ describe("spawn_agent same-cwd concurrency", () => { }); }); -describe("fleetRecords retention cap", () => { +describe("wait mailbox session tombstone and pin", () => { test("many spawned-and-completed workers whose reports are never collected leave memory bounded", async () => { const COUNT = MAX_FLEET_RECORDS + 50; const deps = makeDeps(async () => ({ report: "x".repeat(1000) })); @@ -838,7 +838,7 @@ describe("interrupt_agent unblocks wait_agents", () => { test("soft-interrupt wait collects so a later followup cannot resurrect done", async () => { const sessions = createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(sessions); + const fleetRecords = createFleetMailbox(sessions); const worker = sessions.start({ id: "soft-int", description: "looping", diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index 6e73b13d..ef203639 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -10,9 +10,9 @@ * * Running state is the session store's `WorkerLifecycle`. wait_agents blocks * on that store's `subscribe` raced against a timeout timer, never polling. - * Wait JSON is a projection of stored lifecycle plus a per-install overlay - * (`fleetRecords`): membership, pin, collected, and an optional wait-status - * override. Spawn/resume settlement writes only the session store. + * Wait JSON is a projection of stored lifecycle plus a per-install wait + * mailbox (`FleetMailbox`): membership, pin, collected, and an optional + * wait-status override. Spawn/resume settlement writes only the session store. * * The store's finished-session retention is a TUI display cap (`maxCompleted`, * default 20): `complete()`/`fail()` evict the oldest finished session — @@ -20,7 +20,7 @@ * this because it awaits its own single result before the tool call returns; * here a caller can spawn far more workers than the cap in one turn and only * `wait_agents` them later, so an evicted report would otherwise vanish - * silently. Overlay `register` pins the session (honored by pruneCompleted + * silently. Mailbox `register` pins the session (honored by pruneCompleted * and pruneRetained) until collect unpins. Heavy payloads are still capped at * `MAX_FLEET_RECORDS`: past that, the oldest never-collected pin is compacted * to a tombstone (status only, plus a pointer at `read_agent_trace`). @@ -114,13 +114,12 @@ const RECOVERY_HINT = export const MAX_FLEET_RECORDS = 200; /** - * Per-install wait overlay. Session lifecycle is the source of wait status - * unless this overlay forces interrupted or has frozen a collected result. - * See the module doc for pin/tombstone policy. + * Per-install wait mailbox over the session store. Session lifecycle is the + * source of wait status unless this overlay forces interrupted or has frozen + * a collected result. See the module doc for pin/tombstone policy. */ -class FleetRecords { +class FleetMailbox { private readonly records = new Map(); - private readonly listeners = new Set<() => void>(); private readonly sessions: SubAgentSessionStore; constructor(sessions: SubAgentSessionStore) { @@ -128,17 +127,6 @@ class FleetRecords { sessions?.subscribe(() => this.enforceCap()); } - subscribe(listener: () => void): () => void { - this.listeners.add(listener); - return () => { - this.listeners.delete(listener); - }; - } - - private notify(): void { - for (const listener of this.listeners) listener(); - } - register(id: string): void { const existing = this.records.get(id); const alreadyPinned = @@ -148,16 +136,6 @@ class FleetRecords { this.enforceCap(); } - /** - * Leftover dual-write. Settlement writes the session store; commit 3 deletes this. - */ - resolve(_id: string, _report: string): void {} - - /** - * Leftover dual-write. Settlement writes the session store; commit 3 deletes this. - */ - reject(_id: string, _error: string): void {} - /** * Overlay wait-status override so wait unblocks while the session may still * be running (send_input interrupt:true followup, close_agent teardown). @@ -168,13 +146,11 @@ class FleetRecords { if (existing === undefined) return; if (existing.collected === true) { this.sessions?.wake(); - this.notify(); return; } existing.forceInterrupted = true; this.sessions?.wake(); this.enforceCap(); - this.notify(); } /** @@ -188,7 +164,6 @@ class FleetRecords { if (existing.forceInterrupted !== true) return; delete existing.forceInterrupted; this.sessions?.wake(); - this.notify(); } ids(): string[] { @@ -296,9 +271,9 @@ class FleetRecords { // One overlay per orchestrator install (shared by its spawn_agent and // wait_agents tool instances), not a module singleton — created in // createSpawnAgentTool and threaded to createWaitAgentsTool by the caller. -export type FleetRecordsHandle = FleetRecords; -export function createFleetRecords(sessions: SubAgentSessionStore): FleetRecordsHandle { - return new FleetRecords(sessions); +export type FleetMailboxHandle = FleetMailbox; +export function createFleetMailbox(sessions: SubAgentSessionStore): FleetMailboxHandle { + return new FleetMailbox(sessions); } const SpawnAgentArgs = type({ @@ -403,7 +378,7 @@ export type AgentFleetDeps = SubAgentSandboxDeps & { provider: SubAgentProvider | (() => SubAgentProvider); run: (params: RunSubAgentParams) => Promise; sessions: SubAgentSessionStore; - fleetRecords: FleetRecordsHandle; + fleetRecords: FleetMailboxHandle; /** * Session id of the caller that is mounting this spawn_agent. Nested * orchestrators pass their own worker id so close_agent can walk the tree. @@ -833,10 +808,10 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { interface WaitAgentsDeps { sessions: SubAgentSessionStore; - fleetRecords: FleetRecordsHandle; + fleetRecords: FleetMailboxHandle; } -function isWaitTerminal(id: string, fleetRecords: FleetRecordsHandle): boolean { +function isWaitTerminal(id: string, fleetRecords: FleetMailboxHandle): boolean { const record = fleetRecords.peek(id); return record !== undefined && record.status !== "running"; } @@ -850,7 +825,7 @@ function isWaitTerminal(id: string, fleetRecords: FleetRecordsHandle): boolean { */ async function waitForTerminal( sessions: SubAgentSessionStore, - fleetRecords: FleetRecordsHandle, + fleetRecords: FleetMailboxHandle, targets: readonly string[], timeoutMs: number, mode: "any" | "all", diff --git a/src/subagent/lifecycle-tools.test.ts b/src/subagent/lifecycle-tools.test.ts index 97898852..ad627aea 100644 --- a/src/subagent/lifecycle-tools.test.ts +++ b/src/subagent/lifecycle-tools.test.ts @@ -6,7 +6,7 @@ import { createInterruptAgentTool, createSendInputTool, } from "./lifecycle-tools.js"; -import { createFleetRecords, createWaitAgentsTool } from "./agent-fleet.js"; +import { createFleetMailbox, createWaitAgentsTool } from "./agent-fleet.js"; import { createSubAgentSessionStore } from "./session-store.js"; async function callTool( @@ -54,7 +54,7 @@ describe("close_agent", () => { const closeAgent = createCloseAgentTool({ sessions, - fleetRecords: createFleetRecords(sessions), + fleetRecords: createFleetMailbox(sessions), }); const result = await callTool(closeAgent, { target: parent.id }); @@ -93,7 +93,7 @@ describe("close_agent", () => { describe("resume_agent", () => { test("starts the next turn on a completed retained session and returns immediately", async () => { const sessions = createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(sessions); + const fleetRecords = createFleetMailbox(sessions); const retained = sessions.start({ description: "d", agentId: "a", brief: "b", retained: true }); const history: string[] = ["first task"]; let finish: (reply: string) => void = () => {}; @@ -150,7 +150,7 @@ describe("resume_agent", () => { test("resumes an interrupted retained session without calling close()", async () => { const sessions = createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(sessions); + const fleetRecords = createFleetMailbox(sessions); const worker = sessions.start({ description: "worker", agentId: "a", @@ -203,7 +203,7 @@ describe("resume_agent", () => { test("rejects a closed session and a concurrent resume of a running turn", async () => { const sessions = createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(sessions); + const fleetRecords = createFleetMailbox(sessions); const closed = sessions.start({ description: "closed", agentId: "a", @@ -258,7 +258,7 @@ describe("resume_agent", () => { test("wait_agents collects the resumed turn after resume_agent returns", async () => { const sessions = createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(sessions); + const fleetRecords = createFleetMailbox(sessions); const worker = sessions.start({ description: "worker", agentId: "a", @@ -275,7 +275,6 @@ describe("resume_agent", () => { ); sessions.complete(worker.id, "first report"); fleetRecords.register(worker.id); - fleetRecords.resolve(worker.id, "first report"); const resumeAgent = createResumeAgentTool({ sessions, fleetRecords }); const wait = createWaitAgentsTool({ sessions, fleetRecords }); @@ -309,7 +308,7 @@ describe("interrupt_agent", () => { const interruptAgent = createInterruptAgentTool({ sessions, - fleetRecords: createFleetRecords(sessions), + fleetRecords: createFleetMailbox(sessions), }); if (interruptAgent.kind !== "full") throw new Error("expected full tool"); @@ -406,7 +405,7 @@ describe("send_input", () => { test("rejects completed, interrupted, and closed sessions — steering is in-flight only", async () => { const sessions = createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(sessions); + const fleetRecords = createFleetMailbox(sessions); const sendInput = createSendInputTool({ sessions, fleetRecords }); if (sendInput.kind !== "full") throw new Error("expected full tool"); @@ -570,7 +569,7 @@ describe("nested lifecycle authority", () => { } const interrupt = createInterruptAgentTool({ sessions, - fleetRecords: createFleetRecords(sessions), + fleetRecords: createFleetMailbox(sessions), authority: nestAuthority(sessions, nested.id), }); expect((await callTool(interrupt, { target: child.id })).status).toBe("interrupted"); @@ -596,7 +595,7 @@ describe("nested lifecycle authority", () => { for (const s of [child, sibling]) sessions.registerClose(s.id, async () => {}); const close = createCloseAgentTool({ sessions, - fleetRecords: createFleetRecords(sessions), + fleetRecords: createFleetMailbox(sessions), authority: nestAuthority(sessions, nested.id), }); expect((await callTool(close, { target: child.id })).status).toBe("shutdown"); @@ -633,7 +632,7 @@ describe("nested lifecycle authority", () => { } const resume = createResumeAgentTool({ sessions, - fleetRecords: createFleetRecords(sessions), + fleetRecords: createFleetMailbox(sessions), authority: nestAuthority(sessions, nested.id), }); expect((await callTool(resume, { target: child.id, message: "more" })).status).toBe("running"); diff --git a/src/subagent/lifecycle-tools.ts b/src/subagent/lifecycle-tools.ts index f990f91e..568dd0da 100644 --- a/src/subagent/lifecycle-tools.ts +++ b/src/subagent/lifecycle-tools.ts @@ -13,7 +13,7 @@ import { type } from "arktype"; import type { ToolDefinition, ToolResult } from "@intx/types/runtime"; import { DEFAULT_CLOSE_DEADLINE_MS } from "./dispose.js"; -import type { FleetRecordsHandle } from "./agent-fleet.js"; +import type { FleetMailboxHandle } from "./agent-fleet.js"; import { DEFAULT_MAX_ENTRY_CHARS, type AgentLifecycleStatus, @@ -112,23 +112,23 @@ export interface LifecycleAuthority { export interface LifecycleToolDeps { sessions: SubAgentSessionStore; /** Optional for send_input; close, interrupt, and resume require it. */ - fleetRecords?: FleetRecordsHandle; + fleetRecords?: FleetMailboxHandle; authority?: LifecycleAuthority; } /** close_agent always terminalizes the wait mailbox — no silent skip. */ export type CloseAgentToolDeps = LifecycleToolDeps & { - fleetRecords: FleetRecordsHandle; + fleetRecords: FleetMailboxHandle; }; /** interrupt_agent always terminalizes the wait mailbox — no silent skip. */ export type InterruptAgentToolDeps = LifecycleToolDeps & { - fleetRecords: FleetRecordsHandle; + fleetRecords: FleetMailboxHandle; }; /** resume_agent registers the next turn on the wait mailbox so wait_agents can collect. */ export type ResumeAgentToolDeps = LifecycleToolDeps & { - fleetRecords: FleetRecordsHandle; + fleetRecords: FleetMailboxHandle; }; function gateTarget( diff --git a/src/subagent/run.ts b/src/subagent/run.ts index 28bf2107..19ff475c 100644 --- a/src/subagent/run.ts +++ b/src/subagent/run.ts @@ -116,7 +116,7 @@ import { } from "./dispose.js"; import { createTaskTool } from "./task-tool.js"; import { - createFleetRecords, + createFleetMailbox, createSpawnAgentTool, createWaitAgentsTool, createListAgentsTool, @@ -599,7 +599,7 @@ async function runSubAgentInner( } const nd = params.nestedDispatch; const fleetSessions = nd.sessions ?? createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(fleetSessions); + const fleetRecords = createFleetMailbox(fleetSessions); tools = [ ...tools, createTaskTool({ diff --git a/src/subagent/spawn-agent-worktree.test.ts b/src/subagent/spawn-agent-worktree.test.ts index 52c703b9..32e9e37e 100644 --- a/src/subagent/spawn-agent-worktree.test.ts +++ b/src/subagent/spawn-agent-worktree.test.ts @@ -5,7 +5,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { promisify } from "node:util"; -import { createFleetRecords, createSpawnAgentTool } from "./agent-fleet.js"; +import { createFleetMailbox, createSpawnAgentTool } from "./agent-fleet.js"; import { createSubAgentSessionStore } from "./session-store.js"; import { createPermissionGate } from "../permission/gate.js"; import type { RunSubAgentParams, RunSubAgentResult } from "./types.js"; @@ -105,7 +105,7 @@ describe("spawn_agent worktree isolation", () => { return { report: "done" }; }, sessions, - fleetRecords: createFleetRecords(sessions), + fleetRecords: createFleetMailbox(sessions), }); if (tool.kind !== "full") throw new Error("expected full tool"); const result = await tool.handler( @@ -144,7 +144,7 @@ describe("spawn_agent worktree isolation", () => { return { report: "no" }; }, sessions, - fleetRecords: createFleetRecords(sessions), + fleetRecords: createFleetMailbox(sessions), }); if (tool.kind !== "full") throw new Error("expected full tool"); const result = await tool.handler( @@ -209,7 +209,7 @@ describe("spawn_agent worktree isolation", () => { throw error; }, sessions, - fleetRecords: createFleetRecords(sessions), + fleetRecords: createFleetMailbox(sessions), }); if (tool.kind !== "full") throw new Error("expected full tool"); @@ -260,7 +260,7 @@ describe("spawn_agent worktree isolation", () => { return settle.promise; }, sessions, - fleetRecords: createFleetRecords(sessions), + fleetRecords: createFleetMailbox(sessions), }); if (tool.kind !== "full") throw new Error("expected full tool"); const spawned = await tool.handler( @@ -333,7 +333,7 @@ describe("spawn_agent worktree isolation", () => { return result; }, sessions, - fleetRecords: createFleetRecords(sessions), + fleetRecords: createFleetMailbox(sessions), }); if (tool.kind !== "full") throw new Error("expected full tool"); const spawned = await tool.handler( @@ -393,7 +393,7 @@ describe("spawn_agent worktree isolation", () => { return { report: "## Summary\nSalvaged." }; }, sessions, - fleetRecords: createFleetRecords(sessions), + fleetRecords: createFleetMailbox(sessions), }); if (tool.kind !== "full") throw new Error("expected full tool"); await tool.handler( diff --git a/src/subagent/task-tool.ts b/src/subagent/task-tool.ts index 91e457dc..bd4bdca0 100644 --- a/src/subagent/task-tool.ts +++ b/src/subagent/task-tool.ts @@ -34,12 +34,12 @@ import { import { isCodexProviderName } from "../config/codex-providers.js"; import { DEFAULT_CANCEL_REASON, type SubAgentSessionStore } from "./session-store.js"; import { - createFleetRecords, + createFleetMailbox, createSpawnAgentTool, createWaitAgentsTool, MAX_WAIT_TIMEOUT_MS, type AgentFleetDeps, - type FleetRecordsHandle, + type FleetMailboxHandle, } from "./agent-fleet.js"; import { buildDispatchBrief, type TaskIntent } from "./report.js"; import { appendSubAgentParentHints, type ForcedStopReason } from "./stop-policy.js"; @@ -206,7 +206,7 @@ export type TaskToolDeps = SubAgentSandboxDeps & { // process-wide dependency; omitting it makes dispatch silent. telemetry?: Telemetry; /** Shared with spawn_agent/wait_agents when this task tool is fleet-backed. */ - fleetRecords?: FleetRecordsHandle; + fleetRecords?: FleetMailboxHandle; }; function taskToolResult( @@ -284,7 +284,7 @@ async function runTaskViaFleet(input: { reportFocus: string | undefined; deps: TaskToolDeps; sessions: SubAgentSessionStore; - fleetRecords: FleetRecordsHandle; + fleetRecords: FleetMailboxHandle; }): Promise { const fleetDeps: AgentFleetDeps = { permissionGate: input.deps.permissionGate, @@ -452,7 +452,7 @@ export function createTaskTool(deps: TaskToolDeps): AgentTool { const fleetSessions = deps.sessions; const fleetRecords = deps.fleetRecords ?? - (fleetSessions !== undefined ? createFleetRecords(fleetSessions) : undefined); + (fleetSessions !== undefined ? createFleetMailbox(fleetSessions) : undefined); // Every completed dispatch gets an outcome record — the log otherwise // carries shape and run state but never what the run actually produced. // Tagged with the dispatched child's provider/model/family so diff --git a/src/subagent/task-via-fleet.test.ts b/src/subagent/task-via-fleet.test.ts index 098daa37..e16bf0e9 100644 --- a/src/subagent/task-via-fleet.test.ts +++ b/src/subagent/task-via-fleet.test.ts @@ -1,7 +1,7 @@ import { describe, expect, test } from "bun:test"; import { createTaskTool } from "./task-tool.js"; -import { createFleetRecords } from "./agent-fleet.js"; +import { createFleetMailbox } from "./agent-fleet.js"; import { createSubAgentSessionStore } from "./session-store.js"; import { createPermissionGate } from "../permission/gate.js"; @@ -20,7 +20,7 @@ const provider = { describe("task via spawn_agent + wait_agents", () => { test("a director task with a session store returns the worker report", async () => { const sessions = createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(sessions); + const fleetRecords = createFleetMailbox(sessions); const tool = createTaskTool({ permissionGate: testPermissionGate, cwd: "/tmp", From cfeb388c87cec04ff2a078c5d54433a45614519e Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 31 Aug 2026 16:48:54 -0700 Subject: [PATCH 4/7] Cancel live workers when a headless exec run ends Headless exec finally cancels live sub-agents the same way TUI runtime shutdown does, then closes the primary agent and disposes the toolset. cancelAll is fire-and-forget and does not serialize closeOne. --- docs/ARCHITECTURE.md | 2 +- src/exec/runner.ts | 55 +++++++++++++++++++++++----------- tests/unit/exec/runner.test.ts | 32 ++++++++++++++++++++ 3 files changed, 70 insertions(+), 19 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 5a225c25..99433157 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -100,7 +100,7 @@ In TUI chat mode there is no completion gate — the session stays open across t - Entry: `corbits exec "prompt"` (alias `corbits run`); `loadConfig` sets `command: "exec"` - Streams assistant text deltas to stdout; lifecycle errors to stderr - Shares ChatDirector compaction continuation (`requestContinuation` → content-less deliver after compact) so long runs do not stall post-compact -- Single primary `agent.send(task)` turn; samples run-sink status/error **before** close (close emits `reactor.done` which would clear sticky errors); then closes the agent before draining the stream so the process exits; toolset is always disposed in `finally` +- Single primary `agent.send(task)` turn; samples run-sink status/error **before** close (close emits `reactor.done` which would clear sticky errors); then closes the agent before draining the stream so the process exits; `finally` cancels live sub-agents (`subAgentSessions.cancelAll("Session closed")`, matching TUI runtime-shutdown), closes the agent, and always disposes the toolset - Status: chat sessions rarely emit `reactor.done` before close, so a completed `send()` maps to `done` unless the pre-close run sink holds a real error - Used by `scripts/demo.ts` (mode `exec`) and the capability eval suite (`scripts/eval-capability.ts` / `evals/capability/`) diff --git a/src/exec/runner.ts b/src/exec/runner.ts index dc56d88e..1dcc3d75 100644 --- a/src/exec/runner.ts +++ b/src/exec/runner.ts @@ -45,7 +45,11 @@ import { import { detectLanguageServerAvailable } from "../agent/lsp-availability.js"; import { normalizeToolDefinitionsForProvider } from "../agent/tool-schema-normalize.js"; import { resolveSessionMode, type SessionMode } from "../config/session-mode.js"; -import { createSubAgentSessionStore, type SubAgentProvider } from "../subagent/index.js"; +import { + createSubAgentSessionStore, + type SubAgentProvider, + type SubAgentSessionStore, +} from "../subagent/index.js"; import type { ContextStore, InferenceSource, @@ -116,6 +120,33 @@ export function formatCaughtError(err: unknown): string { return err instanceof Error ? err.message : String(err); } +/** + * Headless analogue of TUI `runtime-shutdown`: abort live workers, then close + * the primary agent and dispose the toolset. `cancelAll` is fire-and-forget — + * it does not serialize `closeOne`. + */ +export async function disposeExecRuntime(args: { + agent: { close: () => Promise } | null; + toolset: { dispose: () => Promise } | null; + subAgentSessions: Pick | null; +}): Promise { + args.subAgentSessions?.cancelAll("Session closed"); + if (args.agent !== null) { + await args.agent.close().catch((err: unknown) => { + logger.debug("agent.close during exec finally failed: {error}", { + error: formatCaughtError(err), + }); + }); + } + if (args.toolset !== null) { + await args.toolset.dispose().catch((err: unknown) => { + logger.debug("toolset.dispose during exec finally failed: {error}", { + error: formatCaughtError(err), + }); + }); + } +} + /** * Exec-primary director overlay. Omit / skywalker keep the product default * (`loadSessionChatPrompt` + advertised session tools). Any other closed-fleet @@ -248,6 +279,7 @@ export async function runExec(config: Config): Promise { let connectedMcp: ConnectedMcpServer[] = []; let agent: Agent | null = null; let toolset: AgentToolset | null = null; + let subAgentSessions: SubAgentSessionStore | null = null; let textOut = ""; let finalized = false; let turnsUsed = 0; @@ -392,7 +424,8 @@ export async function runExec(config: Config): Promise { const liveSubAgentProvider: { current: SubAgentProvider } = { current: buildSubAgentProvider(config), }; - const subAgentSessions = createSubAgentSessionStore(); + const fleetSessions = createSubAgentSessionStore(); + subAgentSessions = fleetSessions; const shellTimeout = shellTimeoutFromSettings(config.settings); const toolWatchdog = toolWatchdogFromSettings(config.settings); const toolAvailability: ToolAvailability = { @@ -447,7 +480,7 @@ export async function runExec(config: Config): Promise { ? { subAgent: { provider: () => liveSubAgentProvider.current, - sessions: subAgentSessions, + sessions: fleetSessions, getWorkdirBase: () => sessionDir(config.cwd, sessionId), onProgress: () => undefined, ...(config.settings !== undefined ? { settings: () => config.settings! } : {}), @@ -888,21 +921,7 @@ export async function runExec(config: Config): Promise { model: config.model, }; } finally { - if (agent !== null) { - await agent.close().catch((err: unknown) => { - logger.debug("agent.close during exec finally failed: {error}", { - error: formatCaughtError(err), - }); - }); - } - // Match TUI: always dispose toolset (MCP clients + posix/plugin resources). - if (toolset !== null) { - await toolset.dispose().catch((err: unknown) => { - logger.debug("toolset.dispose during exec finally failed: {error}", { - error: formatCaughtError(err), - }); - }); - } + await disposeExecRuntime({ agent, toolset, subAgentSessions }); } } diff --git a/tests/unit/exec/runner.test.ts b/tests/unit/exec/runner.test.ts index f475d063..96c451f2 100644 --- a/tests/unit/exec/runner.test.ts +++ b/tests/unit/exec/runner.test.ts @@ -1,11 +1,13 @@ import { describe, expect, test } from "bun:test"; import type { Config } from "../../../src/config/index.js"; import { + disposeExecRuntime, formatCaughtError, resolveExecDirectorOverlay, runExec, } from "../../../src/exec/runner.js"; import { BUILD_TOOLS } from "../../../src/agent/directors/tool-sets.js"; +import { createSubAgentSessionStore } from "../../../src/subagent/session-store.js"; function bareConfig(task: string): Config { // Minimal unconfigured-shaped object is not enough — runExec only needs @@ -54,6 +56,36 @@ describe("runExec", () => { }); }); +describe("disposeExecRuntime", () => { + test("cancels fire-and-forget workers when exec finishes", async () => { + const store = createSubAgentSessionStore(); + const worker = store.start({ description: "bg", agentId: "w", brief: "b" }); + let aborted = 0; + store.registerCancel(worker.id, () => { + aborted += 1; + }); + + const calls: string[] = []; + await disposeExecRuntime({ + agent: { + close: async () => { + calls.push("agent"); + }, + }, + toolset: { + dispose: async () => { + calls.push("toolset"); + }, + }, + subAgentSessions: store, + }); + + expect(aborted).toBe(1); + expect(store.get(worker.id)?.status).toBe("cancelled"); + expect(calls).toEqual(["agent", "toolset"]); + }); +}); + describe("resolveExecDirectorOverlay", () => { test("builder exec primary does not mount task", () => { const overlay = resolveExecDirectorOverlay("builder"); From e08e4b63de2b5db59ca293e84a759aa2c7279129 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 31 Aug 2026 19:15:35 -0700 Subject: [PATCH 5/7] Fix resumed-agent fail teardown and wait pin reuse fail() of a live persisted agent must invoke the registered close so close_agent can recover after a resume followup throw. Mailbox register re-pins on call-id reuse, and a pruned mailbox member is a tombstone instead of eternal running. Wait/task cancel stays interrupted. --- CHANGELOG.md | 5 ++ docs/ARCHITECTURE.md | 2 +- src/subagent/agent-fleet.test.ts | 53 +++++++++++++++++++++ src/subagent/agent-fleet.ts | 24 ++++++++-- src/subagent/lifecycle-tools.test.ts | 38 ++++++++++++++- src/subagent/lifecycle-tools.ts | 2 +- src/subagent/session-store.test.ts | 59 +++++++++++++++++++++-- src/subagent/session-store.ts | 71 ++++++++++++++++++---------- 8 files changed, 215 insertions(+), 39 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e8ad58ce..2d898b58 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,11 @@ parallel copies under `docs/` or `scripts/notes/`. At cut time: rename Dedicated Nordic å in `/model` is treated as that shortcut when Add Provider is offered. +### Changed + +- Cancelling a `task` or `wait_agents` worker reports wait status `interrupted`, + not `failed`. + ### Fixed - Codex ChatGPT subscription sessions no longer show a public-rate dollar diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 99433157..5578d44c 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -226,7 +226,7 @@ Every director package carries a required `tier: SubagentTier` field (`src/agent Enforcement is runtime code at the existing tool-mount point, not prompt wording — this is the fix for four prior mechanisms (`writePaths`, `report.requiredSections`, a `--config` comment, the thrash matcher) that were documented-as-enforced while enforcing nothing: - **Mount-time gate — live today, and fails closed.** `task-tool.ts` resolves the caller's tier at dispatch time — a closed director's `DirectorPackage.tier` — and forwards it as `RunSubAgentParams.orchestratorTier`. `runSubAgent` (`src/subagent/run.ts`) then calls `assertTierMayMountFleetVerb(tier, toolName)` (`src/subagent/authority.ts`) before installing fleet verbs, treating a **missing** `orchestratorTier` as `"leaf"` — deny, not skip. This is the case that matters most: a project-local or plugin `AgentProfile` with `orchestrator: true` is outside the closed director set and is **not** trusted with fleet verbs just because `orchestrator: true` is set — there is no profile-level opt-in today, so the mount always throws `FleetAuthorityError` for a profile-sourced orchestrator. `FLEET_VERBS` in `authority.ts` names the live verbs (`task`, `spawn_agent`, `wait_agents`, `list_agents`, `send_input`, `interrupt_agent`, `close_agent`, `resume_agent`, `read_agent_trace`, `search_agents`) so every mount site inherits the same gate. `list_agents` is the non-blocking mailbox-scoped list of this install's own `spawn_agent` workers (same scope as `wait_agents`); nested orchestrators may mount it. Fleet discovery (`search_agents`) remains Tier 1 only. -- **Subtree authority — wired for addressing verbs.** `assertCanTargetAgent(actor, targetId, nodes)` implements the "root owns its tree; a child manages only its own descendants" rule over the `{id, parentSessionId}` shape `SubAgentSessionStore` already tracks. Production call sites: `read_agent_trace`, `send_input`, `interrupt_agent`, `close_agent`, and `resume_agent`. Nested mounts pass `{actorId, tier, getNodes}` from `run.ts`; a missing `actorId` fails closed. Tier-1 primary omits authority and stays unrestricted. `spawn_agent` records `parentSessionId` on nested workers so `close_agent`'s descendant walk can see them. `wait_agents` with omitted targets waits only on that caller's per-install wait mailbox over the shared session store, not every running session. Wait JSON is a projection of stored lifecycle plus mailbox membership/pin/collected/interrupt override — not a second terminal store. `list_agents` reports that same mailbox without blocking. `interrupt_agent` / `send_input` with `interrupt:true` terminalize the wait mailbox immediately; the soft-interrupt wait path collects so a later followup cannot resurrect an already-observed interrupt. `close_agent` also terminalizes the wait mailbox before teardown. Operator cancel (`cancel` / `cancelAll`) projects wait status `interrupted`. +- **Subtree authority — wired for addressing verbs.** `assertCanTargetAgent(actor, targetId, nodes)` implements the "root owns its tree; a child manages only its own descendants" rule over the `{id, parentSessionId}` shape `SubAgentSessionStore` already tracks. Production call sites: `read_agent_trace`, `send_input`, `interrupt_agent`, `close_agent`, and `resume_agent`. Nested mounts pass `{actorId, tier, getNodes}` from `run.ts`; a missing `actorId` fails closed. Tier-1 primary omits authority and stays unrestricted. `spawn_agent` records `parentSessionId` on nested workers so `close_agent`'s descendant walk can see them. `wait_agents` with omitted targets waits only on that caller's per-install wait mailbox over the shared session store, not every running session. Wait JSON is a projection of stored lifecycle plus mailbox membership/pin/collected/interrupt override — not a second terminal store. `list_agents` reports that same mailbox without blocking. `interrupt_agent` stamps the session `interrupted`; wait JSON projects that stored lifecycle and does not write a mailbox overlay. `send_input` with `interrupt:true` sets the mailbox interrupt overlay so wait unblocks while a queued followup may already be running. The wait path collects a terminal status so a later followup cannot resurrect an already-observed interrupt. `close_agent` terminalizes the wait mailbox before teardown. Operator cancel (`cancel` / `cancelAll`) projects wait status `interrupted`. - `task()` remains the deprecated fused spawn+wait fallback. `spawn_agent` + `wait_agents` is the supported parallel path. The tier check still gates which packages may mount any fleet verb. #### Closed director fleet (`src/agent/directors/`) diff --git a/src/subagent/agent-fleet.test.ts b/src/subagent/agent-fleet.test.ts index 42040c44..701c2b40 100644 --- a/src/subagent/agent-fleet.test.ts +++ b/src/subagent/agent-fleet.test.ts @@ -415,6 +415,59 @@ describe("wait mailbox session tombstone and pin", () => { expect(results[0]!.report).toBeUndefined(); expect(results[0]!.hint).toContain("read_agent_trace"); }); + + test("spawn_agent call.id reuse still pins the new session", async () => { + let t = 0; + const sessions = createSubAgentSessionStore({ + maxCompleted: 1, + now: () => ++t, + }); + const deps = makeDeps(async () => ({ report: "ok" }), { sessions }); + const spawn = createSpawnAgentTool(deps); + const wait = createWaitAgentsTool({ sessions, fleetRecords: deps.fleetRecords }); + if (spawn.kind !== "full") throw new Error("expected full tool"); + const args = { description: "job", prompt: "do it", intent: "explore" }; + const signal = new AbortController().signal; + + await spawn.handler({ id: "reuse-id", name: "spawn_agent", arguments: args }, signal); + await new Promise((resolve) => setTimeout(resolve, 20)); + await spawn.handler({ id: "reuse-id", name: "spawn_agent", arguments: args }, signal); + await new Promise((resolve) => setTimeout(resolve, 20)); + + const extra1 = sessions.start({ description: "flood-1", agentId: "a", brief: "b" }); + sessions.complete(extra1.id, "flood-1"); + const extra2 = sessions.start({ description: "flood-2", agentId: "a", brief: "b" }); + sessions.complete(extra2.id, "flood-2"); + + expect(sessions.get("reuse-id")).toBeDefined(); + const waited = await callTool(wait, { targets: ["reuse-id"], timeout_ms: 1000 }); + expect(waited.timed_out).toBe(false); + const results = waited.results as { status: string }[]; + expect(results[0]!.status).toBe("done"); + }); + + test("wait on a pruned mailbox member is tombstone not eternal running", () => { + let t = 0; + const sessions = createSubAgentSessionStore({ + maxCompleted: 1, + now: () => ++t, + }); + const mailbox = createFleetMailbox(sessions); + sessions.start({ id: "reuse", description: "old", agentId: "a", brief: "b" }); + mailbox.register("reuse"); + sessions.complete("reuse", "old report"); + sessions.start({ id: "reuse", description: "new", agentId: "a", brief: "b" }); + sessions.complete("reuse", "new report"); + const extra = sessions.start({ description: "other", agentId: "a", brief: "b" }); + sessions.complete(extra.id, "other"); + const extra2 = sessions.start({ description: "prune", agentId: "a", brief: "b" }); + sessions.complete(extra2.id, "prune"); + expect(sessions.get("reuse")).toBeUndefined(); + const snap = mailbox.peek("reuse"); + expect(snap?.status).not.toBe("running"); + expect(snap?.tombstoned).toBe(true); + expect(snap?.hint).toContain("read_agent_trace"); + }); }); describe("spawn_agent parentage", () => { diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index ef203639..3badc2fb 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -129,10 +129,11 @@ class FleetMailbox { register(id: string): void { const existing = this.records.get(id); - const alreadyPinned = - existing !== undefined && existing.collected !== true && existing.pinHeld === true; + // start() drops pinCounts on call-id reuse. Re-pin whenever the overlay + // thought it still held a pin, so wait cannot desync against an empty map. + if (existing?.pinHeld === true) this.sessions.unpin(id); this.records.set(id, { pinHeld: true }); - if (!alreadyPinned) this.sessions?.pin(id); + this.sessions.pin(id); this.enforceCap(); } @@ -212,7 +213,7 @@ class FleetMailbox { private projectedStatus(id: string, overlay: FleetOverlay): WaitJSONStatus { if (overlay.frozenStatus !== undefined) return overlay.frozenStatus; if (overlay.forceInterrupted === true) return "interrupted"; - return this.sessionWaitStatus(id) ?? "running"; + return this.sessionWaitStatus(id) ?? "interrupted"; } snapshot(id: string): FleetRecord { @@ -220,8 +221,21 @@ class FleetMailbox { if (overlay === undefined) { return { status: "running" }; } + const session = this.sessions.get(id); + if ( + session === undefined && + overlay.tombstoned !== true && + overlay.frozenStatus === undefined + ) { + overlay.tombstoned = true; + overlay.hint = RECOVERY_HINT; + overlay.frozenStatus = "interrupted"; + if (overlay.pinHeld === true) { + overlay.pinHeld = false; + this.sessions.unpin(id); + } + } const status = this.projectedStatus(id, overlay); - const session = this.sessions?.get(id); const sessionWait = this.sessionWaitStatus(id); const payload = overlay.tombstoned !== true && session !== undefined && sessionWait === status diff --git a/src/subagent/lifecycle-tools.test.ts b/src/subagent/lifecycle-tools.test.ts index ad627aea..e62cec99 100644 --- a/src/subagent/lifecycle-tools.test.ts +++ b/src/subagent/lifecycle-tools.test.ts @@ -132,9 +132,9 @@ describe("resume_agent", () => { finish("done, history now 2 turns"); await new Promise((resolve) => setTimeout(resolve, 0)); - expect(sessions.get(retained.id)?.lifecycleStatus).toBe("completed"); + expect(sessions.get(retained.id)?.lifecycleStatus).toBe("interrupted"); expect(sessions.get(retained.id)?.id).toBe(retained.id); - expect(sessions.get(retained.id)?.report).toBe("done, history now 2 turns"); + expect(sessions.get(retained.id)?.report).toBe("## Summary\nDone."); if (resumeAgent.kind !== "full") throw new Error("expected full tool"); const rejected = await resumeAgent.handler( @@ -298,6 +298,40 @@ describe("resume_agent", () => { expect(results[0]!.status).toBe("done"); expect(results[0]!.report).toBe("second report"); }); + + test("resume followup rejection invokes close; close_agent tears down leftover", async () => { + const sessions = createSubAgentSessionStore(); + const fleetRecords = createFleetMailbox(sessions); + const worker = sessions.start({ + description: "worker", + agentId: "a", + brief: "b", + retained: true, + }); + let closeCalls = 0; + sessions.registerClose(worker.id, async () => { + closeCalls++; + }); + sessions.registerFollowup(worker.id, async () => { + throw new Error("send failed"); + }); + sessions.complete(worker.id, "first report"); + + const resumeAgent = createResumeAgentTool({ sessions, fleetRecords }); + const closeAgent = createCloseAgentTool({ sessions, fleetRecords }); + const resumed = await callTool(resumeAgent, { target: worker.id, message: "again" }); + expect(resumed.status).toBe("running"); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(closeCalls).toBe(1); + expect(sessions.get(worker.id)?.lifecycle.state).toBe("failed"); + + const started = Date.now(); + const closed = await callTool(closeAgent, { target: worker.id }); + expect(Date.now() - started).toBeLessThan(1000); + expect(closed.status).toBe("shutdown"); + expect(sessions.get(worker.id)?.lifecycle.state).toBe("failed"); + expect(closeCalls).toBe(1); + }); }); describe("interrupt_agent", () => { diff --git a/src/subagent/lifecycle-tools.ts b/src/subagent/lifecycle-tools.ts index 568dd0da..ac787b6e 100644 --- a/src/subagent/lifecycle-tools.ts +++ b/src/subagent/lifecycle-tools.ts @@ -121,7 +121,7 @@ export type CloseAgentToolDeps = LifecycleToolDeps & { fleetRecords: FleetMailboxHandle; }; -/** interrupt_agent always terminalizes the wait mailbox — no silent skip. */ +/** interrupt_agent stamps session interrupted; wait JSON projects that lifecycle. */ export type InterruptAgentToolDeps = LifecycleToolDeps & { fleetRecords: FleetMailboxHandle; }; diff --git a/src/subagent/session-store.test.ts b/src/subagent/session-store.test.ts index 8fe6f315..c7abc6d3 100644 --- a/src/subagent/session-store.test.ts +++ b/src/subagent/session-store.test.ts @@ -365,8 +365,9 @@ describe("CL-6943 reusable worker sessions", () => { finish("later"); await new Promise((resolve) => setTimeout(resolve, 0)); - expect(store.get(session.id)?.status).toBe("done"); - expect(store.get(session.id)?.lifecycleStatus).toBe("completed"); + expect(store.get(session.id)?.status).toBe("running"); + expect(store.get(session.id)?.lifecycleStatus).toBe("interrupted"); + expect(store.get(session.id)?.report).toBe("## Summary\nDone."); }); test("rejected followup restores strip status so interrupt_agent fails closed", async () => { @@ -698,9 +699,9 @@ describe("interrupt stamps finishedAt once", () => { t = 5000; finish("later"); await new Promise((resolve) => setTimeout(resolve, 0)); - expect(store.get(session.id)?.status).toBe("done"); - expect(store.get(session.id)?.lifecycleStatus).toBe("completed"); - expect(store.get(session.id)?.finishedAt).toBe(5000); + expect(store.get(session.id)?.status).toBe("running"); + expect(store.get(session.id)?.lifecycleStatus).toBe("interrupted"); + expect(store.get(session.id)?.finishedAt).toBe(4000); }); test("a follow-up turn keeps the lane live past the linger window until it completes", async () => { @@ -756,6 +757,54 @@ describe("interrupt stamps finishedAt once", () => { }); describe("CL-7269 one stored worker lifecycle", () => { + test("complete() after interrupt_agent does not overwrite interrupted", () => { + const store = createSubAgentSessionStore(); + const session = store.start({ description: "d", agentId: "a", brief: "b", retained: true }); + store.markRunning(session.id); + store.registerInterrupt(session.id, () => {}); + expect(store.interruptOne(session.id).ok).toBe(true); + store.complete(session.id, "late original send"); + const after = store.get(session.id); + expect(after?.lifecycle.state).toBe("interrupted"); + expect(after?.lifecycleStatus).toBe("interrupted"); + expect(after?.report).toBeUndefined(); + }); + + test("fail() of a live persisted agent invokes the registered close", async () => { + const store = createSubAgentSessionStore(); + const session = store.start({ description: "d", agentId: "a", brief: "b", retained: true }); + store.markRunning(session.id); + let closeCalls = 0; + store.registerClose(session.id, async () => { + closeCalls++; + }); + store.fail(session.id, "send failed"); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(closeCalls).toBe(1); + expect(store.get(session.id)?.lifecycle.state).toBe("failed"); + const started = Date.now(); + const status = await store.closeOne(session.id, 5000); + expect(Date.now() - started).toBeLessThan(200); + expect(status).toBe("shutdown"); + expect(store.get(session.id)?.lifecycle.state).toBe("failed"); + expect(closeCalls).toBe(1); + }); + + test("closeOne still tears down a leftover close handle after fail()", async () => { + const store = createSubAgentSessionStore(); + const session = store.start({ description: "d", agentId: "a", brief: "b", retained: true }); + store.markRunning(session.id); + let closeCalls = 0; + store.fail(session.id, "send failed"); + store.registerClose(session.id, async () => { + closeCalls++; + }); + const status = await store.closeOne(session.id, 5000); + expect(status).toBe("shutdown"); + expect(closeCalls).toBe(1); + expect(store.get(session.id)?.lifecycle.state).toBe("failed"); + }); + test("fail() stores failed, projects strip failed and verb shutdown", () => { const store = createSubAgentSessionStore(); const session = store.start({ description: "d", agentId: "a", brief: "b" }); diff --git a/src/subagent/session-store.ts b/src/subagent/session-store.ts index a5510782..383969ea 100644 --- a/src/subagent/session-store.ts +++ b/src/subagent/session-store.ts @@ -623,12 +623,16 @@ export function createSubAgentSessionStore( }; const check = (): void => { const session = sessions.get(id); - if (session === undefined || isAlreadyClosed(session.lifecycle)) { + if (session === undefined) { finish(undefined); return; } const close = closeHandles.get(id); - if (close !== undefined) finish(close); + if (close !== undefined) { + finish(close); + return; + } + if (isAlreadyClosed(session.lifecycle)) finish(undefined); }; listeners.add(listener); const timer = setTimeout(() => finish(closeHandles.get(id)), deadlineMs); @@ -694,7 +698,8 @@ export function createSubAgentSessionStore( if ( still.lifecycle.state === "shutdown" || still.lifecycle.state === "cancelled" || - still.lifecycle.state === "failed" + still.lifecycle.state === "failed" || + still.lifecycle.state === "interrupted" ) { runInFlight.delete(id); return; @@ -936,11 +941,18 @@ export function createSubAgentSessionStore( // always passes this flag explicitly (see its call site). const agentRetained = opts?.agentRetained ?? true; mutate(id, (session) => { - // Cancel wins races: a late complete after operator cancel must not - // resurrect the session as done. - if (!isLiveStrip(session.lifecycle) || session.lifecycle.state === "cancelled") return; - // Interrupted is still strip-live; a settling run may complete. Operator - // cancel is not live for this path because state is cancelled. + // Cancel and interrupt_agent win races: a late complete must not + // resurrect the session as done. Interrupted is still strip-live + // (linger), so it needs an explicit check. Salvage bodies attach via + // attachReport without changing state. send_input interrupt:true + // followup goes through beginFollowupTurn (running) and still completes. + if ( + !isLiveStrip(session.lifecycle) || + session.lifecycle.state === "cancelled" || + session.lifecycle.state === "interrupted" + ) { + return; + } session.lifecycle = { state: "completed", report }; if (!agentRetained) session.retained = false; session.finishedAt = now(); @@ -961,14 +973,11 @@ export function createSubAgentSessionStore( fail(id: string, error: string): void { mutate(id, (session) => { if (!isLiveStrip(session.lifecycle) || session.lifecycle.state === "cancelled") return; - // A thrown run always tears down its agent in run.ts's finally - // (persist only skips teardown on a clean success) — so there is - // nothing left to resume here, and retained no longer applies. - // CL-7001: a deadline/cancel salvage does NOT throw — it returns a - // report through the same success path a clean completion uses, so - // it never reaches this function. complete() carries the equivalent - // "agent was actually disposed" check for that case via its - // agentRetained flag; this function only ever needed to cover throws. + // Spawn-path throws already dispose in run.ts's finally. Resume of a + // persisted agent does not: the live close handle is the only teardown. + // Invoke it fire-and-forget (same as prune/evict) without marking + // shutdown — fail stays fail, and an already-disposed spawn close is + // best-effort idempotent. session.lifecycle = { state: "failed", error }; session.retained = false; session.finishedAt = now(); @@ -978,9 +987,8 @@ export function createSubAgentSessionStore( kind: "report", content: capText(`Error: ${error}`, maxEntryChars), }); - cancelHandles.delete(id); - closeHandles.delete(id); runInFlight.delete(id); + releaseHandles(id); pruneCompleted(); }); }, @@ -1016,8 +1024,11 @@ export function createSubAgentSessionStore( if (evicted.has(id)) return "shutdown"; return "not_found"; } - if (isAlreadyClosed(session.lifecycle)) return projectLifecycleStatus(session.lifecycle); let close = closeHandles.get(id); + const alreadyClosed = isAlreadyClosed(session.lifecycle); + if (alreadyClosed && close === undefined) { + return projectLifecycleStatus(session.lifecycle); + } if (close === undefined) { // CL-7001: close_agent landed in the setup window — the session // exists but createAgentWithLiveToolDispatch hasn't finished and @@ -1029,16 +1040,14 @@ export function createSubAgentSessionStore( close = await waitForCloseHandle(id, deadlineMs); const stillHere = sessions.get(id); if (stillHere === undefined) return "not_found"; - if (isAlreadyClosed(stillHere.lifecycle)) { - return projectLifecycleStatus(stillHere.lifecycle); - } if (close === undefined) { - // Never became closeable within the deadline: report the honest - // in-progress status rather than a false "shutdown" — the caller - // can retry, and this session is still findable to retry against. + // Never became closeable within the deadline, or fail() already + // released the handle: report the honest stored status rather + // than a false "shutdown". return projectLifecycleStatus(stillHere.lifecycle); } } + const keepFailed = isAlreadyClosed((sessions.get(id) ?? session).lifecycle); closeHandles.delete(id); // Bounded here too, defense-in-depth against a caller-registered // close that does not honor its own deadline argument — a wedged @@ -1051,6 +1060,18 @@ export function createSubAgentSessionStore( }), new Promise((resolve) => setTimeout(resolve, deadlineMs)), ]); + if (keepFailed) { + // fail() already stamped failed; invoke leftover teardown without + // rewriting that to shutdown. + cancelHandles.delete(id); + interruptHandles.delete(id); + followupHandles.delete(id); + deliverHandles.delete(id); + runInFlight.delete(id); + pruneCompleted(); + const after = sessions.get(id); + return after === undefined ? "not_found" : projectLifecycleStatus(after.lifecycle); + } mutate(id, (s) => { const wasLive = isLiveStrip(s.lifecycle); const error = s.error ?? (wasLive ? "Closed by close_agent" : undefined); From b9243519f98b6e6746f6eba14d8d06ffcb327ca7 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 31 Aug 2026 20:21:21 -0700 Subject: [PATCH 6/7] Freeze pruned mailbox wait status at last known result A completed worker pruned from the session store was wait-reported as interrupted because snapshot hard-froze missing sessions. Capture the live wait projection and retention tombstone so wait keeps done/failed, and only unknown missing sessions stay interrupted. --- src/subagent/agent-fleet.test.ts | 56 +++++++++++++++++++++++++++++--- src/subagent/agent-fleet.ts | 51 ++++++++++++++++++++++++++--- src/subagent/session-store.ts | 10 ++++++ 3 files changed, 108 insertions(+), 9 deletions(-) diff --git a/src/subagent/agent-fleet.test.ts b/src/subagent/agent-fleet.test.ts index 701c2b40..05d9a79e 100644 --- a/src/subagent/agent-fleet.test.ts +++ b/src/subagent/agent-fleet.test.ts @@ -62,6 +62,33 @@ function makeDeps( }; } +function waitUntilMailboxTerminal( + mailbox: ReturnType, + sessions: ReturnType, + id: string, +): Promise { + return new Promise((resolve) => { + const done = (): boolean => { + const snap = mailbox.peek(id); + return snap !== undefined && snap.status !== "running"; + }; + if (done()) { + resolve(); + return; + } + const unsub = sessions.subscribe(() => { + if (done()) { + unsub(); + resolve(); + } + }); + if (done()) { + unsub(); + resolve(); + } + }); +} + async function callToolRaw( tool: ReturnType | ReturnType, args: Record, @@ -422,7 +449,16 @@ describe("wait mailbox session tombstone and pin", () => { maxCompleted: 1, now: () => ++t, }); - const deps = makeDeps(async () => ({ report: "ok" }), { sessions }); + const firstRun = deferred(); + const secondRun = deferred(); + let calls = 0; + const deps = makeDeps( + async () => { + calls += 1; + return (calls === 1 ? firstRun : secondRun).promise; + }, + { sessions }, + ); const spawn = createSpawnAgentTool(deps); const wait = createWaitAgentsTool({ sessions, fleetRecords: deps.fleetRecords }); if (spawn.kind !== "full") throw new Error("expected full tool"); @@ -430,9 +466,12 @@ describe("wait mailbox session tombstone and pin", () => { const signal = new AbortController().signal; await spawn.handler({ id: "reuse-id", name: "spawn_agent", arguments: args }, signal); - await new Promise((resolve) => setTimeout(resolve, 20)); + firstRun.resolve({ report: "ok" }); + await callTool(wait, { targets: ["reuse-id"], timeout_ms: 5000 }); + await spawn.handler({ id: "reuse-id", name: "spawn_agent", arguments: args }, signal); - await new Promise((resolve) => setTimeout(resolve, 20)); + secondRun.resolve({ report: "ok" }); + await waitUntilMailboxTerminal(deps.fleetRecords, sessions, "reuse-id"); const extra1 = sessions.start({ description: "flood-1", agentId: "a", brief: "b" }); sessions.complete(extra1.id, "flood-1"); @@ -464,10 +503,19 @@ describe("wait mailbox session tombstone and pin", () => { sessions.complete(extra2.id, "prune"); expect(sessions.get("reuse")).toBeUndefined(); const snap = mailbox.peek("reuse"); - expect(snap?.status).not.toBe("running"); + expect(snap?.status).toBe("done"); expect(snap?.tombstoned).toBe(true); expect(snap?.hint).toContain("read_agent_trace"); }); + + test("wait on a mailbox member with no session history is interrupted", () => { + const sessions = createSubAgentSessionStore(); + const mailbox = createFleetMailbox(sessions); + mailbox.register("ghost"); + const snap = mailbox.peek("ghost"); + expect(snap?.status).toBe("interrupted"); + expect(snap?.tombstoned).toBe(true); + }); }); describe("spawn_agent parentage", () => { diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index 3badc2fb..26ea1e19 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -59,7 +59,11 @@ import type { Settings } from "../config/settings.js"; import { resolveEffortForRole } from "../provider/reasoning-effort.js"; import { isCodexProviderName } from "../config/codex-providers.js"; import { buildDispatchBrief, type TaskIntent } from "./report.js"; -import { DEFAULT_CANCEL_REASON, type SubAgentSessionStore } from "./session-store.js"; +import { + DEFAULT_CANCEL_REASON, + type AgentLifecycleStatus, + type SubAgentSessionStore, +} from "./session-store.js"; import { projectWaitStatus, type WaitJSONStatus } from "./lifecycle.js"; import type { NestedDispatchDeps, @@ -103,6 +107,8 @@ interface FleetOverlay { forceInterrupted?: boolean; /** Frozen wait status after collect. Later session completed must not resurrect this mailbox. */ frozenStatus?: WaitJSONStatus; + /** Last wait projection seen while the session still existed. */ + lastWaitStatus?: WaitJSONStatus; tombstoned?: boolean; hint?: string; } @@ -110,6 +116,14 @@ interface FleetOverlay { const RECOVERY_HINT = "Report evicted to bound fleet memory; recover full detail via read_agent_trace(agent_id)."; +function waitStatusFromVerbLifecycle( + status: AgentLifecycleStatus | undefined, +): WaitJSONStatus | undefined { + if (status === "completed") return "done"; + if (status === "interrupted" || status === "shutdown") return "interrupted"; + return undefined; +} + /** Payload cap: uncollected pinned terminal records still holding a report. */ export const MAX_FLEET_RECORDS = 200; @@ -124,7 +138,10 @@ class FleetMailbox { constructor(sessions: SubAgentSessionStore) { this.sessions = sessions; - sessions?.subscribe(() => this.enforceCap()); + sessions?.subscribe(() => { + this.rememberLiveWaitStatuses(); + this.enforceCap(); + }); } register(id: string): void { @@ -132,7 +149,11 @@ class FleetMailbox { // start() drops pinCounts on call-id reuse. Re-pin whenever the overlay // thought it still held a pin, so wait cannot desync against an empty map. if (existing?.pinHeld === true) this.sessions.unpin(id); - this.records.set(id, { pinHeld: true }); + const wait = this.sessionWaitStatus(id); + this.records.set(id, { + pinHeld: true, + ...(wait !== undefined ? { lastWaitStatus: wait } : {}), + }); this.sessions.pin(id); this.enforceCap(); } @@ -204,6 +225,17 @@ class FleetMailbox { return this.snapshot(id); } + private rememberLiveWaitStatuses(): void { + for (const [id, overlay] of this.records) { + const wait = this.sessionWaitStatus(id); + if (wait !== undefined) overlay.lastWaitStatus = wait; + } + } + + private waitStatusFromEvicted(id: string): WaitJSONStatus | undefined { + return waitStatusFromVerbLifecycle(this.sessions.evictedLifecycle(id)); + } + private sessionWaitStatus(id: string): WaitJSONStatus | undefined { const session = this.sessions?.get(id); if (session === undefined) return undefined; @@ -213,7 +245,16 @@ class FleetMailbox { private projectedStatus(id: string, overlay: FleetOverlay): WaitJSONStatus { if (overlay.frozenStatus !== undefined) return overlay.frozenStatus; if (overlay.forceInterrupted === true) return "interrupted"; - return this.sessionWaitStatus(id) ?? "interrupted"; + const live = this.sessionWaitStatus(id); + if (live !== undefined) { + overlay.lastWaitStatus = live; + return live; + } + const last = overlay.lastWaitStatus; + if (last !== undefined && last !== "running") return last; + const evicted = this.waitStatusFromEvicted(id); + if (evicted !== undefined && evicted !== "running") return evicted; + return "interrupted"; } snapshot(id: string): FleetRecord { @@ -229,7 +270,7 @@ class FleetMailbox { ) { overlay.tombstoned = true; overlay.hint = RECOVERY_HINT; - overlay.frozenStatus = "interrupted"; + overlay.frozenStatus = this.projectedStatus(id, overlay); if (overlay.pinHeld === true) { overlay.pinHeld = false; this.sessions.unpin(id); diff --git a/src/subagent/session-store.ts b/src/subagent/session-store.ts index 383969ea..75000935 100644 --- a/src/subagent/session-store.ts +++ b/src/subagent/session-store.ts @@ -158,6 +158,12 @@ export interface SubAgentSessionStoreOptions { export interface SubAgentSessionStore { list(): readonly SubAgentSession[]; get(id: string): SubAgentSession | undefined; + /** + * Verb lifecycle of a session dropped by pruneRetained, if a tombstone remains. + * `get` does not surface these — they exist so wait/resume can recover a + * terminal status instead of treating the id as never-seen. + */ + evictedLifecycle(id: string): AgentLifecycleStatus | undefined; // Running + recent completed, newest first — surface for the Agents strip. listForStrip(): readonly SubAgentSession[]; start(input: StartSessionInput): SubAgentSession; @@ -738,6 +744,10 @@ export function createSubAgentSessionStore( return session === undefined ? undefined : snapshotOf(session); }, + evictedLifecycle(id: string): AgentLifecycleStatus | undefined { + return evicted.get(id)?.lifecycleStatus; + }, + listForStrip(): readonly SubAgentSession[] { return [...sessions.values()].map(snapshotOf).sort((a, b) => { // Running first, then by startedAt descending. From 9957775006e1c46489e5c5710d038dc8ba4b42a7 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 31 Aug 2026 21:35:32 -0700 Subject: [PATCH 7/7] Pass deliver into host tests after steer routing landed --- src/tui/runner-host.test.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/tui/runner-host.test.ts b/src/tui/runner-host.test.ts index e630039a..601dddf5 100644 --- a/src/tui/runner-host.test.ts +++ b/src/tui/runner-host.test.ts @@ -146,6 +146,7 @@ describe("mountRunnerHost chrome wiring", () => { eventEmitter: new EventEmitter(), send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: {}, onModelSelect: () => {}, commands: () => commands, @@ -432,6 +433,7 @@ describe("mountRunnerHost model picker", () => { eventEmitter: new EventEmitter(), send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: { xai: { models: ["grok-4"] } }, onModelSelect: () => {}, onConnectProvider: () => {}, @@ -463,6 +465,7 @@ describe("mountRunnerHost model picker", () => { eventEmitter: new EventEmitter(), send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: { xai: { models: ["grok-4"] } }, onModelSelect: () => {}, commands: [],