From 7284f9bfab1736387d6d04f1be423fe47f1bdb86 Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Sat, 1 Aug 2026 01:25:59 +0800 Subject: [PATCH 1/2] fix: sweep orphaned sub-agents at turn boundaries and on abort paths The provider's `subagent_stop` event was the ONLY path that removed an entry from Session's #subagents map. That event originates in the Claude SDK's SubagentStop hook, which cannot fire once the query is aborted -- and interrupt, setModel, rotate and switchProvider all abort it via #abortController.abort(). Every such abort therefore orphaned each in-flight sub-agent permanently, with nothing anywhere reconciling the map afterwards. Three consequences: - subagentSnapshot (feeding /who and toInfo().subagents) only ever grew. The displayed sub-agent count climbed turn after turn and never came back down, which is how this was noticed. - #subagents and #subagentRegistrations grew unbounded for the lifetime of a long-lived session. - Worst: each orphan kept a LIVE delegated ZeroID token. Revocation is meant to ride the sub-agent's own stop; instead it fell through to deactivateSessionAgent's cascade at session destroy, so a dead sub-agent's credential stayed valid as long as the session lived. That quietly weakens the per-agent revocation guarantee. Adds #sweepStaleSubagents, which revokes and drops whatever remains. A Task sub-agent cannot outlive the turn that spawned it, so it runs at turn_done as the principled backstop, plus on the two abort paths that leave the session alive: interrupt() and #teardownProvider(). It is idempotent and returns immediately on an empty map (the common case), and double-revocation is free because deactivateSubagent no-ops on an id it already dropped -- so a trailing subagent_stop racing a sweep costs nothing. Not swept at destroy(): deactivateSessionAgent already cascades over the session's sub-agent keys there, and the Session object is discarded. Tests: 6 of the 8 new cases fail against the pre-fix code, covering the turn-end sweep, non-accumulation across turns, both abort paths, the identity revocation, and the info_update broadcast so clients see the count drop. Co-Authored-By: Claude Opus 5 (1M context) --- src/daemon/session.ts | 58 +++++ src/tests/session-subagent-lifecycle.test.ts | 247 +++++++++++++++++++ 2 files changed, 305 insertions(+) create mode 100644 src/tests/session-subagent-lifecycle.test.ts diff --git a/src/daemon/session.ts b/src/daemon/session.ts index 33626c0..d0d1477 100644 --- a/src/daemon/session.ts +++ b/src/daemon/session.ts @@ -1840,6 +1840,52 @@ export class Session { this.#broadcastInfoUpdate(); } + /** + * Drop every sub-agent still registered and revoke its ZeroID identity. + * + * A Task sub-agent cannot outlive the turn that spawned it, so once that turn + * ends — or the provider it was running under goes away — any surviving entry + * is stale by definition. + * + * Before this existed the ONLY cleanup path was the provider's + * `subagent_stop` event, which originates in the SDK's SubagentStop hook. + * That hook cannot fire when the query is aborted mid-turn (interrupt, + * setModel, rotate, provider switch — all of which call + * `#abortController.abort()`), so each abort permanently orphaned every + * in-flight sub-agent. Three consequences, all of which this fixes: + * + * 1. `subagentSnapshot` (→ `/who`, `toInfo().subagents`) only ever grew — + * the reported count climbed across turns and never came back down. + * 2. `#subagents` + `#subagentRegistrations` grew unbounded for the + * lifetime of a long-lived session. + * 3. Worst: each orphan kept a LIVE delegated ZeroID token. Revocation is + * supposed to ride the sub-agent's own stop; instead it waited for + * `deactivateSessionAgent`'s cascade at session destroy, so a dead + * sub-agent's credential stayed valid for as long as the session lived. + * + * Idempotent and cheap — a no-op when the map is empty, which is the common + * case. Double-revocation is safe: `deactivateSubagent` no-ops on an id it + * has already dropped, so a `subagent_stop` arriving after a sweep (or a + * sweep racing a trailing stop) costs nothing. + */ + #sweepStaleSubagents(reason: string): void { + if (this.#subagents.size === 0) return; + const orphaned = [...this.#subagents.keys()]; + for (const agentId of orphaned) { + // Fire-and-forget, matching the subagent_stop path: revocation must never + // block a turn boundary, and deactivateSubagent logs its own failures. + void this.#identityManager?.deactivateSubagent(this.id, agentId); + this.#subagentRegistrations.delete(agentId); + this.#subagents.delete(agentId); + } + // Worth a line: a non-empty sweep means a SubagentStop never arrived, which + // is expected on abort but would otherwise be invisible. + console.log( + `[codeoid/session ${this.id}] swept ${orphaned.length} stale sub-agent(s) at ${reason}`, + ); + this.#broadcastInfoUpdate(); + } + async #teardownProvider(): Promise { // Capture before nulling: provider.teardown() may trigger onRecoveryNeeded, // which installs a new #eventConsumerTask. Awaiting the snapshot drains @@ -1851,6 +1897,10 @@ export class Session { this.#eventConsumerTask = null; await this.#provider.teardown(); try { await taskToAwait; } catch { /* consumer handles its own errors */ } + // teardown() aborts the SDK query, so any sub-agent still in flight will + // never get its SubagentStop hook. This is the setModel / rotate / + // switchProvider path — the session survives, so the orphans would too. + this.#sweepStaleSubagents("provider teardown"); // The drained consumer's `finally` skips its own idle reset here: we nulled // #activeRun above, so its run-ownership guard (`#activeRun === run`) is // false. Without this, tearing a provider down mid-turn (setModel / rotate) @@ -1917,12 +1967,16 @@ export class Session { if (run) { try { await run.interrupt(); + // Interrupting kills the turn, so every sub-agent it spawned is done + // whether or not the SDK got to run their stop hooks. + this.#sweepStaleSubagents("interrupt"); if (this.#status !== "error") this.#setStatus("idle"); return; } catch { // fall through to hard abort } } + this.#sweepStaleSubagents("interrupt"); if (this.#status !== "error") this.#setStatus("idle"); } @@ -3762,6 +3816,10 @@ export class Session { case "turn_done": { this.#accumulator.handleEvent(event); this.#recordTurnFromResult(event.result); + // The turn is over, so no Task sub-agent it spawned can still be alive. + // Normally every one of them already emitted subagent_stop and this is a + // no-op; it's the backstop for the ones whose hook never fired. + this.#sweepStaleSubagents("turn end"); // Hook seam: observe-only (git-checkpoint per turn, usage export). this.#hookBus?.emit("after_turn", this.#hookContext(), { result: event.result, diff --git a/src/tests/session-subagent-lifecycle.test.ts b/src/tests/session-subagent-lifecycle.test.ts new file mode 100644 index 0000000..baf9998 --- /dev/null +++ b/src/tests/session-subagent-lifecycle.test.ts @@ -0,0 +1,247 @@ +/** + * Sub-agent lifecycle cleanup — the orphan/leak regression. + * + * `subagent_stop` (the SDK's SubagentStop hook) used to be the ONLY path that + * removed an entry from Session's `#subagents` map. That hook cannot fire when + * the SDK query is aborted mid-turn — interrupt, setModel, rotate, provider + * switch all call `#abortController.abort()` — so every abort permanently + * orphaned each in-flight sub-agent: + * + * - `subagentSnapshot` only ever grew (the reported count never came down) + * - `#subagents` / `#subagentRegistrations` grew for the session's lifetime + * - each orphan kept a LIVE delegated ZeroID token until session destroy + * + * `#sweepStaleSubagents` now runs at every turn boundary and on the abort + * paths. These tests pin all three consequences. + */ + +import { describe, it, expect, beforeEach, afterEach } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { randomUUID } from "node:crypto"; +import { Store } from "../daemon/store.js"; +import { TranscriptStore } from "../daemon/transcript.js"; +import { Session, type AttachedClient } from "../daemon/session.js"; +import { MockSessionProvider, mockResult } from "../daemon/providers/mock/session-provider.js"; +import { ProviderRegistry } from "../daemon/providers/registry.js"; +import type { ProviderEvent } from "../daemon/providers/interface.js"; +import type { AgentIdentityManager } from "../daemon/agent-identity.js"; +import type { AuthContext, DaemonMessage } from "../protocol/types.js"; +import { ALL_SCOPES } from "../protocol/scopes.js"; + +const AUTH: AuthContext = { + sub: "user:subagent-test", + scopes: [...ALL_SCOPES] as AuthContext["scopes"], + delegationDepth: 0, + accountId: "acc-sa", + projectId: "proj-sa", +}; + +let tmp: string; +let store: Store; +let transcriptStore: TranscriptStore; + +beforeEach(() => { + tmp = mkdtempSync(join(tmpdir(), "codeoid-subagent-")); + store = new Store(join(tmp, "codeoid.db")); + transcriptStore = new TranscriptStore(join(tmp, "transcripts")); +}); + +afterEach(async () => { + await new Promise((r) => setTimeout(r, 50)); + try { await transcriptStore.flush(); } catch {} + try { store.close(); } catch {} + try { rmSync(tmp, { recursive: true, force: true }); } catch {} +}); + +const spawn = (agentId: string, agentType = "explore"): ProviderEvent => + ({ type: "subagent_start", agentId, agentType }); +const stop = (agentId: string): ProviderEvent => ({ type: "subagent_stop", agentId }); +const done = (): ProviderEvent => ({ type: "turn_done", result: mockResult() }); + +/** + * Records deactivation calls so tests can assert the ZeroID token for an + * orphaned sub-agent actually gets revoked (the leak that mattered most). + */ +function stubIdentityManager() { + const deactivatedSubagents: Array<{ sessionId: string; agentId: string }> = []; + const mgr = { + async registerSessionAgent(sessionId: string) { + return { wimseUri: `spiffe://test/session/${sessionId}`, token: "test-token" }; + }, + async registerSubagent(_sessionId: string, agentId: string, agentType: string) { + return { wimseUri: `spiffe://test/subagent/${agentType}/${agentId}` }; + }, + async deactivateSubagent(sessionId: string, agentId: string) { + deactivatedSubagents.push({ sessionId, agentId }); + }, + async deactivateSessionAgent() {}, + } as unknown as AgentIdentityManager; + return { mgr, deactivatedSubagents }; +} + +function makeSession( + script: ProviderEvent[][], + opts: { stall?: boolean; identityManager?: AgentIdentityManager } = {}, +): { session: Session; provider: MockSessionProvider } { + const id = randomUUID(); + const provider = new MockSessionProvider("mock", script, { stall: opts.stall ?? false }); + const registry = new ProviderRegistry("mock"); + registry.register({ id: "mock", displayName: "mock", create: () => provider }); + store.createSession({ + id, + name: "subagent-test", + workdir: tmp, + status: "idle", + createdBy: AUTH.sub, + createdAt: new Date().toISOString(), + attachedClients: 0, + accountId: AUTH.accountId, + projectId: AUTH.projectId, + }); + const session = new Session({ + name: "subagent-test", + workdir: tmp, + auth: AUTH, + store, + transcriptStore, + existingId: id, + providers: registry, + providerId: "mock", + ...(opts.identityManager ? { identityManager: opts.identityManager } : {}), + }); + return { session, provider }; +} + +function recordingClient(): AttachedClient & { received: DaemonMessage[] } { + const received: DaemonMessage[] = []; + return { id: randomUUID(), auth: AUTH, received, send: (m) => { received.push(m); } }; +} + +async function waitFor(cond: () => boolean, ms = 2000): Promise { + const deadline = Date.now() + ms; + while (!cond()) { + if (Date.now() > deadline) throw new Error("waitFor timed out"); + await new Promise((r) => setTimeout(r, 10)); + } +} + +describe("sub-agent cleanup at turn boundaries", () => { + it("sweeps sub-agents whose stop hook never fired when the turn ends", async () => { + // Three spawns, zero stops — what an aborted or hook-dropping turn leaves. + const { session } = makeSession([[spawn("a1"), spawn("a2"), spawn("a3"), done()]]); + session.attach(recordingClient()); + + await session.send("go", AUTH); + await waitFor(() => session.status === "idle"); + + // Pre-fix this was 3, and stayed 3 while later turns added more. + expect(session.subagentSnapshot).toHaveLength(0); + }); + + it("does not accumulate across turns", async () => { + const { session } = makeSession([ + [spawn("t1-a"), spawn("t1-b"), done()], + [spawn("t2-a"), spawn("t2-b"), done()], + ]); + session.attach(recordingClient()); + + await session.send("turn one", AUTH); + await waitFor(() => session.status === "idle"); + expect(session.subagentSnapshot).toHaveLength(0); + + await session.send("turn two", AUTH); + await waitFor(() => session.status === "idle"); + // The reported symptom: 2 → 4 → 6 … as turns went by. + expect(session.subagentSnapshot).toHaveLength(0); + }); + + it("leaves the normal stop path intact (no double-handling)", async () => { + const { session } = makeSession([[spawn("clean"), stop("clean"), done()]]); + session.attach(recordingClient()); + + await session.send("go", AUTH); + await waitFor(() => session.status === "idle"); + + expect(session.subagentSnapshot).toHaveLength(0); + }); + + it("revokes the ZeroID identity of a swept sub-agent", async () => { + const { mgr, deactivatedSubagents } = stubIdentityManager(); + const { session } = makeSession([[spawn("orphan"), done()]], { identityManager: mgr }); + session.attach(recordingClient()); + + await session.send("go", AUTH); + await waitFor(() => session.status === "idle"); + await waitFor(() => deactivatedSubagents.length > 0); + + // The leak that mattered: pre-fix this token stayed live until the whole + // session was destroyed. + expect(deactivatedSubagents.map((d) => d.agentId)).toEqual(["orphan"]); + }); + + it("revokes each orphan exactly once even when a stop also arrives", async () => { + const { mgr, deactivatedSubagents } = stubIdentityManager(); + const { session } = makeSession([[spawn("both"), stop("both"), done()]], { + identityManager: mgr, + }); + session.attach(recordingClient()); + + await session.send("go", AUTH); + await waitFor(() => session.status === "idle"); + await waitFor(() => deactivatedSubagents.length > 0); + + // subagent_stop revoked it and removed the entry, so the sweep finds + // nothing left to revoke. + expect(deactivatedSubagents).toHaveLength(1); + }); +}); + +describe("sub-agent cleanup on abort paths", () => { + it("sweeps on interrupt — the path where SubagentStop provably cannot fire", async () => { + // stall: emit the spawns, then never close the queue and never emit + // turn_done. This is a turn still in flight, exactly as when a user + // interrupts mid-Task. + const { session } = makeSession([[spawn("live-1"), spawn("live-2")]], { stall: true }); + session.attach(recordingClient()); + + await session.send("go", AUTH); + await waitFor(() => session.subagentSnapshot.length === 2); + + await session.interrupt(AUTH); + + expect(session.subagentSnapshot).toHaveLength(0); + }); + + it("sweeps on provider teardown (setModel / rotate / switch)", async () => { + const { session } = makeSession([[spawn("live-1"), spawn("live-2")]], { stall: true }); + session.attach(recordingClient()); + + await session.send("go", AUTH); + await waitFor(() => session.subagentSnapshot.length === 2); + + // setModel tears the provider down mid-turn (session.ts:2639), aborting the + // SDK query — so no SubagentStop can arrive for the two live sub-agents. + await session.setModel("haiku", undefined, AUTH); + + expect(session.subagentSnapshot).toHaveLength(0); + }); + + it("broadcasts info_update when a sweep actually removes something", async () => { + const { session } = makeSession([[spawn("live-1")]], { stall: true }); + const client = recordingClient(); + session.attach(client); + + await session.send("go", AUTH); + await waitFor(() => session.subagentSnapshot.length === 1); + const before = client.received.filter((m) => m.type === "session.info_update").length; + + await session.interrupt(AUTH); + + // Clients must learn the count dropped, not just the daemon. + expect( + client.received.filter((m) => m.type === "session.info_update").length, + ).toBeGreaterThan(before); + }); +}); From e1524b2912ee38953e378d11eabc1318ac6a9a14 Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Sun, 2 Aug 2026 00:45:06 +0800 Subject: [PATCH 2/2] fix: stop losing turn-lifecycle events instead of only reconciling after The sweep added earlier reconciled dangling sub-agents but never asked why they dangled. Root-causing that turned up two defects, one of which is the "model answered but the spinner keeps spinning" report. 1. Mid-turn accounting counted pushes that produce no turn_done. #pendingMidTurnCount exists so #consumeEvents can absorb the intermediate turn_done the SDK emits when a mid-turn push starts a NEW query. It was incremented for every mid-turn push, including "later" -- which merges into the running turn and starts no query at all (ClaudeProvider: shouldQuery = priority !== "later"). Any client sending with an explicit "later" priority while the session was working therefore left the counter one too high, and the consumer swallowed the turn's REAL terminal turn_done as if it were an intermediate boundary: it re-asserted "thinking" and continue'd, waiting for a turn_done that would never be emitted. That is the stuck spinner. It also stranded the turn's sub-agents and tools, because the loop never exited and so neither the boundary flush nor the consumer's finally ever ran. Recovery came only from the 5-minute stall watchdog or the next send. Now only querying pushes are counted. 2. The provider's event channel dropped events silently. ClaudeProvider.#emit was `try { queue?.push(e) } catch {}` -- three silent losses in one expression: a null queue, a closed queue, and a full queue, all indistinguishable and none logged. A lost subagent_stop left its sub-agent dangling with a live delegated ZeroID token; a lost tool_complete stranded the status at tool_running. Nothing said so. #emit now reports the disposition, and id-keyed lifecycle events (subagent_stop, tool_complete) are buffered and replayed into the next turn instead of vanishing -- their handlers are idempotent on unknown ids, so a late duplicate costs nothing. turn_done and streamed text are deliberately not carried: a stale turn_done would end the next turn as it began. The buffer is bounded and the log is deduped per loop generation. There was also a fourth, invisible loss: after the consumer broke on turn_done, the turn queue stayed OPEN and unread until the next turn replaced it, so late events were accepted into a queue nobody would drain -- push succeeded, so no error, no log. TurnRun.endTurn() (optional, called once from the consumer's finally) closes it, converting that into the observable carryover path. 3. Sweep moved to where reconciliation already lives. #completeActiveTools has always run in the consumer's finally precisely because provider events can be lost; sub-agents were simply never added to the same backstop. The sweep now sits beside it, covering every exit path, plus the mid-turn continuation branch -- which continue's without dispatching to #handleProviderEvent and so was missed entirely by the previous turn_done-case sweep. The abort-path sweeps stay as defence in depth for interrupt/teardown, where hooks provably never fire. Tests: 5 new cases on top of the existing 8. The three covering these changes fail against the previous commit -- the stuck-spinner case by timing out, which is the bug exactly. Co-Authored-By: Claude Opus 5 (1M context) --- src/daemon/providers/claude/index.ts | 114 ++++++++++++++++- src/daemon/providers/interface.ts | 16 +++ src/daemon/providers/mock/session-provider.ts | 28 ++++ src/daemon/session.ts | 40 +++++- src/tests/session-subagent-lifecycle.test.ts | 120 +++++++++++++++++- 5 files changed, 306 insertions(+), 12 deletions(-) diff --git a/src/daemon/providers/claude/index.ts b/src/daemon/providers/claude/index.ts index 8395918..3f6861f 100644 --- a/src/daemon/providers/claude/index.ts +++ b/src/daemon/providers/claude/index.ts @@ -119,6 +119,25 @@ export interface ClaudeProviderInit { onRecoveryNeeded?: (content: string) => void; } +/** + * Events safe to replay into a LATER turn when they can't be delivered now. + * + * Both are addressed by an id the Session tracks (`agentId`, `sdkToolUseId`) + * and both handlers no-op on an unknown id, so arriving late is harmless — + * whereas losing them leaves a dangling sub-agent (with a live delegated token) + * or a tool stuck "running". Deliberately excludes `turn_done` (a stale one + * would end the next turn the moment it starts) and streamed text (it would + * corrupt the next turn's transcript). + */ +export const CARRYOVER_EVENT_TYPES: ReadonlySet = new Set([ + "subagent_stop", + "tool_complete", +]); + +/** Ceiling on buffered undeliverable events, so a pathological loop can't turn + * a delivery failure into unbounded memory growth. */ +export const MAX_CARRYOVER_EVENTS = 100; + // ── ClaudeProvider ──────────────────────────────────────────────────────────── export class ClaudeProvider implements SessionProvider { @@ -164,6 +183,12 @@ export class ClaudeProvider implements SessionProvider { // Long-running event queue — closed only when the SDK loop ends. // turn_done events are emitted as regular items; Session decides when to stop. #currentTurnQueue: AsyncQueue | null = null; + /** Id-keyed lifecycle events that arrived with no live queue, replayed into + * the next turn. See #handleUndeliverable. */ + #carryover: ProviderEvent[] = []; + /** Dedupe keys for the undeliverable-event log, so a persistent loss reports + * once per loop generation instead of once per event. */ + #undeliverableLogged = new Set(); /** * Mutable callback — Session updates this before each runTurn() call to @@ -226,6 +251,28 @@ export class ClaudeProvider implements SessionProvider { const turnQueue = new AsyncQueue(); this.#currentTurnQueue = turnQueue; + // Replay lifecycle events that arrived while no queue was live — a + // SubagentStop hook resolving after the previous turn ended, a trailing + // tool_result. Their handlers are id-keyed and idempotent, so a duplicate + // costs nothing; losing them is what left sub-agents dangling and tools + // stuck "running" for the rest of the session. + if (this.#carryover.length > 0) { + const replay = this.#carryover; + this.#carryover = []; + console.error( + `[claude-provider ${this.#claudeCodeSessionId.slice(0, 8)}] replaying ${replay.length} carried-over event(s) into the new turn`, + ); + for (const ev of replay) { + try { + turnQueue.push(ev); + } catch { + // A brand-new queue should never reject; if it somehow does, the + // boundary sweep is still the backstop. + break; + } + } + } + let userMessage = opts.userMessage; if (this.#pendingHistorySeed && userMessage) { // Post-switch seeding: this fresh Claude Code session has never seen @@ -256,8 +303,24 @@ export class ClaudeProvider implements SessionProvider { // A "later" mid-turn injection is meant to MERGE into the running turn // (no new query); "now"/"next" should query. Only the primary prompt // (above, via the default) always queries. + // + // NOTE for callers: only the querying variants produce an additional + // turn_done. Session mirrors this when deciding whether to expect an + // intermediate turn boundary — counting a "later" push there makes it + // swallow the turn's terminal turn_done and hang at "thinking". this.#pushSDKMessage(content, priority, priority !== "later"); }, + endTurn: () => { + // The consumer for THIS turn has stopped reading. Close its queue so a + // late event surfaces through #handleUndeliverable (logged, and buffered + // if it's id-keyed) instead of being pushed into a queue nobody drains. + // Generation/identity guarded: a rebuilt loop or a newer turn may + // already own #currentTurnQueue, and closing that would kill a live turn. + if (this.#currentTurnQueue === turnQueue) { + this.#currentTurnQueue.close(); + this.#currentTurnQueue = null; + } + }, }; } @@ -652,11 +715,54 @@ export class ClaudeProvider implements SessionProvider { /** Push a ProviderEvent to the active per-turn queue. */ #emit(event: ProviderEvent): void { - try { - this.#currentTurnQueue?.push(event); - } catch { - // Queue may be closed if the turn ended early — ignore. + const queue = this.#currentTurnQueue; + if (queue) { + try { + queue.push(event); + return; + } catch (err) { + this.#handleUndeliverable(event, err instanceof Error ? err.name : "unknown"); + return; + } + } + this.#handleUndeliverable(event, "no-queue"); + } + + /** + * A provider event had nowhere to go — the turn queue was closed (the + * consumer finished and called endTurn), full, or already nulled after the + * SDK loop ended. + * + * This used to be a bare `catch {}`, which is how the whole class of bugs + * stayed invisible: a lost `subagent_stop` left its sub-agent dangling for the + * life of the session (and its delegated ZeroID token live), and a lost + * `tool_complete` stranded the status at `tool_running`, with nothing logged + * to explain either. + * + * Id-keyed lifecycle events are buffered and replayed into the next turn's + * queue, where their handlers are idempotent no-ops if the boundary sweep + * already reconciled them. Everything else is logged only — replaying a stale + * `turn_done` would end the next turn the instant it began, and stale text + * would corrupt its transcript. + */ + #handleUndeliverable(event: ProviderEvent, reason: string): void { + const carryable = CARRYOVER_EVENT_TYPES.has(event.type); + if (carryable && this.#carryover.length < MAX_CARRYOVER_EVENTS) { + this.#carryover.push(event); } + // One line per (loop generation, event type, reason): enough to diagnose a + // recurring loss, not enough to flood a long session. + const key = `${this.#loopGeneration}:${event.type}:${reason}`; + if (this.#undeliverableLogged.has(key)) return; + this.#undeliverableLogged.add(key); + const disposition = carryable + ? this.#carryover.length < MAX_CARRYOVER_EVENTS + ? "buffered for the next turn" + : "DROPPED (carryover full)" + : "dropped"; + console.error( + `[claude-provider ${this.#claudeCodeSessionId.slice(0, 8)}] "${event.type}" undeliverable (${reason}) — ${disposition}`, + ); } /** diff --git a/src/daemon/providers/interface.ts b/src/daemon/providers/interface.ts index d8e84e0..96c7a9a 100644 --- a/src/daemon/providers/interface.ts +++ b/src/daemon/providers/interface.ts @@ -248,6 +248,22 @@ export interface TurnRun { interrupt(): Promise; /** Push a message mid-turn (ClaudeProvider only). */ pushMidTurn?(content: string, priority: "now" | "next" | "later"): void; + /** + * Signal that the consumer has stopped reading `events` — called from + * Session's turn-exit path, exactly once per run. + * + * Keep-warm providers hold one queue per turn but only close it when the NEXT + * turn replaces it. Between those points the queue is open with nobody + * draining it, so a late event (a `SubagentStop` hook resolving after the + * result message, a trailing tool_result) is accepted and then discarded + * unread — the silent loss that leaves sub-agents dangling and tools stuck + * "running". Closing here turns that into an observable, recoverable case: + * the provider sees the closed queue and can buffer or log instead. + * + * Optional and best-effort — providers with no per-turn queue omit it, and it + * must never throw into the consumer's finally. + */ + endTurn?(): void; } // ── ModelInfo ───────────────────────────────────────────────────────────────── diff --git a/src/daemon/providers/mock/session-provider.ts b/src/daemon/providers/mock/session-provider.ts index 485121d..f1963a1 100644 --- a/src/daemon/providers/mock/session-provider.ts +++ b/src/daemon/providers/mock/session-provider.ts @@ -107,6 +107,27 @@ export class MockSessionProvider implements SessionProvider { /** Every pushMidTurn injection observed — inspect in tests. */ readonly midTurnPushes: Array<{ content: string; priority: string }> = []; + /** Times Session called `TurnRun.endTurn()` — the turn-exit signal. */ + endTurnCount = 0; + + /** + * Push an event into the LIVE turn queue from a test: the deterministic + * stand-in for "the SDK emitted this later in the turn". Needed to model a + * terminal turn_done arriving after a mid-turn push (the pendingMidTurnCount + * hang), which a static script can't express. Returns false when the turn's + * queue is already gone or closed. + */ + emitLive(event: ProviderEvent): boolean { + const q = this.#currentQueue; + if (!q) return false; + try { + q.push(event); + return true; + } catch { + return false; + } + } + constructor( id = "mock-session", script: ProviderEvent[][] = [], @@ -199,6 +220,13 @@ export class MockSessionProvider implements SessionProvider { interrupt: async () => { queue.close(); // idempotent — safe to call even if already closed }, + endTurn: () => { + // Mirror ClaudeProvider: the consumer has stopped reading, so close the + // queue rather than leaving it open and undrained. + this.endTurnCount++; + queue.close(); + if (this.#currentQueue === queue) this.#currentQueue = null; + }, }; if (this.#midTurn) { run.pushMidTurn = (content: string, priority: string) => { diff --git a/src/daemon/session.ts b/src/daemon/session.ts index d0d1477..9ab23b9 100644 --- a/src/daemon/session.ts +++ b/src/daemon/session.ts @@ -1747,8 +1747,21 @@ export class Session { this.#persistAndBuffer(midTurnMsg); this.#broadcastRaw(midTurnMsg); this.#accumulator.pushUserTurn(effectivePrompt); - this.#pendingMidTurnCount++; - this.#activeRun.pushMidTurn(effectivePrompt, effectivePriority ?? "now"); + // Count ONLY pushes that start a new query, because only those produce an + // extra turn_done for #consumeEvents to absorb. A "later" push merges into + // the running turn (ClaudeProvider: `shouldQuery = priority !== "later"`), + // so counting it made the consumer treat the turn's REAL terminal + // turn_done as an intermediate boundary: it decremented, re-asserted + // "thinking", and `continue`d — waiting forever for a turn_done that was + // never going to be emitted. + // + // That is the stuck spinner. The model has answered, but the session sits + // at "thinking" until the 5-minute stall watchdog fires or the next send + // closes the queue. It also strands the turn's sub-agents and tools: + // neither the mid-turn flush nor the consumer's finally reconciles them, + // because the loop never exits. + if (effectivePriority !== "later") this.#pendingMidTurnCount++; + this.#activeRun.pushMidTurn(effectivePrompt, effectivePriority); // Keep waiting_approval visible — the approval is still pending and // every frontend keys its approval bar off it; the queued text is // consumed after the user answers. @@ -3363,6 +3376,11 @@ export class Session { this.#recordTurnFromResult(event.result); // Flush per-turn accumulators so the continuation turn starts clean. this.#completeActiveTools(); + // Same boundary, same reasoning: the sub-agents of the partial turn + // are done with it. This branch `continue`s without dispatching to + // #handleProviderEvent, so it is the only place that can reconcile + // them for an absorbed mid-turn boundary. + this.#sweepStaleSubagents("mid-turn boundary"); this.#flushActiveAssistant(); this.#finalizeActiveThinking(); this.#chunker?.onTurnEnd(); @@ -3394,6 +3412,17 @@ export class Session { } finally { this.#pendingMidTurnCount = 0; // safety: reset on any exit path this.#completeActiveTools(); + // Sub-agent reconciliation belongs beside the tool reconciliation: this + // finally is the one path every turn exit goes through — clean turn_done, + // error, stall recovery, ownership loss. #completeActiveTools has always + // existed here because provider events can be lost; sub-agents were simply + // never added to the same backstop. + this.#sweepStaleSubagents("turn exit"); + // Tell the provider this turn's stream has no reader anymore. Without it + // the queue stays open and unconsumed until the NEXT turn replaces it, so + // a late event is silently buffered into a queue nobody will ever drain. + // Closing converts that invisible loss into the provider's carryover path. + try { run.endTurn?.(); } catch { /* best-effort */ } this.#flushActiveAssistant(); this.#finalizeActiveThinking(); this.#chunker?.onTurnEnd(); @@ -3816,10 +3845,9 @@ export class Session { case "turn_done": { this.#accumulator.handleEvent(event); this.#recordTurnFromResult(event.result); - // The turn is over, so no Task sub-agent it spawned can still be alive. - // Normally every one of them already emitted subagent_stop and this is a - // no-op; it's the backstop for the ones whose hook never fired. - this.#sweepStaleSubagents("turn end"); + // No sweep here: a terminal turn_done breaks #consumeEvents, whose + // finally reconciles sub-agents alongside tools. Sweeping here too would + // just be a redundant pass a few statements earlier. // Hook seam: observe-only (git-checkpoint per turn, usage export). this.#hookBus?.emit("after_turn", this.#hookContext(), { result: event.result, diff --git a/src/tests/session-subagent-lifecycle.test.ts b/src/tests/session-subagent-lifecycle.test.ts index baf9998..9b67a6f 100644 --- a/src/tests/session-subagent-lifecycle.test.ts +++ b/src/tests/session-subagent-lifecycle.test.ts @@ -25,6 +25,10 @@ import { TranscriptStore } from "../daemon/transcript.js"; import { Session, type AttachedClient } from "../daemon/session.js"; import { MockSessionProvider, mockResult } from "../daemon/providers/mock/session-provider.js"; import { ProviderRegistry } from "../daemon/providers/registry.js"; +import { + CARRYOVER_EVENT_TYPES, + MAX_CARRYOVER_EVENTS, +} from "../daemon/providers/claude/index.js"; import type { ProviderEvent } from "../daemon/providers/interface.js"; import type { AgentIdentityManager } from "../daemon/agent-identity.js"; import type { AuthContext, DaemonMessage } from "../protocol/types.js"; @@ -83,10 +87,13 @@ function stubIdentityManager() { function makeSession( script: ProviderEvent[][], - opts: { stall?: boolean; identityManager?: AgentIdentityManager } = {}, + opts: { stall?: boolean; midTurn?: boolean; identityManager?: AgentIdentityManager } = {}, ): { session: Session; provider: MockSessionProvider } { const id = randomUUID(); - const provider = new MockSessionProvider("mock", script, { stall: opts.stall ?? false }); + const provider = new MockSessionProvider("mock", script, { + stall: opts.stall ?? false, + midTurn: opts.midTurn ?? false, + }); const registry = new ProviderRegistry("mock"); registry.register({ id: "mock", displayName: "mock", create: () => provider }); store.createSession({ @@ -245,3 +252,112 @@ describe("sub-agent cleanup on abort paths", () => { ).toBeGreaterThan(before); }); }); + +describe("turn-boundary accounting (the stuck-'thinking' root cause)", () => { + /** + * `#pendingMidTurnCount` exists so the consumer can absorb the intermediate + * turn_done the SDK emits when a mid-turn push starts a NEW query. It used to + * be incremented for every mid-turn push — including "later", which merges + * into the running turn and produces no extra turn_done at all + * (ClaudeProvider: `shouldQuery = priority !== "later"`). + * + * The consequence was the turn's real terminal turn_done being swallowed as + * an intermediate boundary: status re-asserted to "thinking", loop + * `continue`d, and the session hung until the 5-minute stall watchdog or the + * next send. The turn's sub-agents were stranded with it, because neither the + * boundary flush nor the consumer's finally ever ran. + */ + it("a 'later' mid-turn push does not make the session swallow its terminal turn_done", async () => { + const { session, provider } = makeSession([[spawn("sa-1")]], { stall: true, midTurn: true }); + session.attach(recordingClient()); + + await session.send("start work", AUTH); + await waitFor(() => session.subagentSnapshot.length === 1); + expect(session.status).not.toBe("idle"); + + // "later" merges into the running turn — no second query, no second turn_done. + await session.send("also consider this", AUTH, undefined, "later"); + expect(provider.midTurnPushes).toEqual([ + { content: expect.stringContaining("also consider this"), priority: "later" }, + ]); + + // The one and only turn_done for this turn. + expect(provider.emitLive({ type: "turn_done", result: mockResult() })).toBe(true); + + // Pre-fix this timed out: status stayed "thinking" forever. + await waitFor(() => session.status === "idle"); + // And the turn's sub-agent is reconciled, because the consumer actually exited. + expect(session.subagentSnapshot).toHaveLength(0); + }); + + it("still absorbs the intermediate turn_done for a querying ('now') push", async () => { + const { session, provider } = makeSession([[spawn("sa-1")]], { stall: true, midTurn: true }); + session.attach(recordingClient()); + + await session.send("start work", AUTH); + await waitFor(() => session.subagentSnapshot.length === 1); + + // "now" DOES start another query, so two turn_dones are expected. + await session.send("actually, do this instead", AUTH, undefined, "now"); + expect(provider.midTurnPushes[0]!.priority).toBe("now"); + + // First turn_done is the intermediate boundary — absorbed, session keeps working. + provider.emitLive({ type: "turn_done", result: mockResult() }); + await new Promise((r) => setTimeout(r, 50)); + expect(session.status).not.toBe("idle"); + + // The continuation's terminal turn_done ends it. + provider.emitLive({ type: "turn_done", result: mockResult() }); + await waitFor(() => session.status === "idle"); + }); + + it("reconciles sub-agents at an absorbed mid-turn boundary", async () => { + const { session, provider } = makeSession([[spawn("boundary-1")]], { + stall: true, + midTurn: true, + }); + session.attach(recordingClient()); + + await session.send("start work", AUTH); + await waitFor(() => session.subagentSnapshot.length === 1); + + await session.send("steer", AUTH, undefined, "now"); + provider.emitLive({ type: "turn_done", result: mockResult() }); // intermediate + + // The mid-turn branch `continue`s without dispatching to + // #handleProviderEvent, so this boundary is the only place that can clear it. + await waitFor(() => session.subagentSnapshot.length === 0); + }); + + it("signals turn exit to the provider so late events can't land in an undrained queue", async () => { + const { session, provider } = makeSession([[spawn("x"), done()]]); + session.attach(recordingClient()); + + await session.send("go", AUTH); + await waitFor(() => session.status === "idle"); + + expect(provider.endTurnCount).toBe(1); + // Queue closed: a late event now surfaces through the provider's + // undeliverable path instead of being buffered where nobody reads it. + expect(provider.emitLive(stop("x"))).toBe(false); + }); +}); + +describe("undeliverable-event carryover policy", () => { + it("carries id-keyed lifecycle events and refuses turn-scoped ones", () => { + // Safe to replay late: id-keyed, and their handlers no-op on unknown ids. + expect(CARRYOVER_EVENT_TYPES.has("subagent_stop")).toBe(true); + expect(CARRYOVER_EVENT_TYPES.has("tool_complete")).toBe(true); + + // Never replay: a stale turn_done would end the next turn the instant it + // started, and stale text would corrupt its transcript. + expect(CARRYOVER_EVENT_TYPES.has("turn_done")).toBe(false); + expect(CARRYOVER_EVENT_TYPES.has("text_delta")).toBe(false); + expect(CARRYOVER_EVENT_TYPES.has("text_done")).toBe(false); + expect(CARRYOVER_EVENT_TYPES.has("error")).toBe(false); + + // Bounded, so a delivery failure can't become unbounded memory growth. + expect(MAX_CARRYOVER_EVENTS).toBeGreaterThan(0); + expect(MAX_CARRYOVER_EVENTS).toBeLessThanOrEqual(1000); + }); +});