From 7a70f82739e085014cad878f08513cdef7b7fe16 Mon Sep 17 00:00:00 2001 From: Ben Brandt Date: Mon, 13 Jul 2026 17:16:11 +0200 Subject: [PATCH] fix: hold a turn open while its background subagents are still live (#870) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem `session/prompt` settled at the turn's terminal `result`, but a turn that launched an **async subagent** is not done at that point. Everything after settlement — the subagent's streamed tool calls, the model's promised task-notification followup summary, and any `session/request_permission` the subagent raises — landed **outside any turn**. Some clients stops consuming updates at the prompt response, so the output was silently lost and the permission request deadlocked on an RPC nobody answers. Reported as #864 (no signal for the post-turn activity), #865 (background tasks not trackable), and #866 (end-to-end field report incl. the permission deadlock). ## What I verified before designing this Driving the adapter over real ACP stdio against the bundled CLI (2.1.206) with `emitRawSDKMessages` on showed that the SDK's trailing `session_state_changed: idle` **does not wait for background agents** — it follows the user result immediately, while the subagent is still running, and the session then cycles: ``` user result → idle → (subagent works) → task_notification → followup turn → result(origin: task-notification) → idle ``` So the old #463-era "settle at idle" design would **not** fix this on current CLIs (and plain re-adopting it would still trade away #773). The hold has to be keyed on the live-subagent set and span those idle cycles. ## Fix A user turn whose terminal result arrives while subagents **it spawned** are still live stores its outcome on `Turn.deferredSettle` instead of settling. The prompt stays pending, so all of the subagent's output and permission requests arrive inside a live turn. It settles with the stored stop reason/usage: - at the **task-notification followup's terminal result** — the promised summary has fully streamed by then (the common case), or - at an **idle with none of its subagents left**, when no followup comes, or - earlier, via `session/cancel` or the next prompt's echo hand-off — a long-running subagent never holds the prompt hostage, and the hand-off reports the recorded stop reason instead of rewriting it to `end_turn`. Tracking: `Turn.spawnedTaskIds` ∩ `session.liveSubagentTasks`, fed by `task_started` (only entries with `subagent_type`; background shells are deliberately excluded — a dev server outlives every turn), pruned at `task_notification`/terminal `task_updated`, and reconciled against `background_tasks_changed`'s replace-semantics payload so a lost bookend can't wedge the hold. A spawn during the held-open window (an agent chain) extends the hold naturally. Turns that spawned nothing (or whose subagents finished in-turn) still settle at their result — the #773 responsiveness is untouched. ## Live verification (adapter on real stdio, CLI 2.1.206, temp cwd) Prompt: *"launch a background agent to read the two files and create summary.txt, then end your turn saying you'll report back"*: ``` [ 15.72] raw: result success ← user result; subagent live → HELD [ 15.72] raw: session_state_changed idle ← immediate idle survives the hold [ 17.50] update: tool_call Read hello.txt ← subagent work, in-turn [ 21.39] update: tool_call Write summary.txt ← in-turn [ 24.17] raw: task_notification completed [ 34.35] update: agent_message_chunk "The background agent finished. It created summary.txt…" ← promised summary, in-turn [ 35.70] raw: result success (origin task-notification) [ 35.70] PROMPT RESOLVED end_turn, usage = user turn's own tokens ``` No-regression check: a plain `"say hello"` prompt resolves the instant its result lands (2.09s, zero added latency). ## Hardening from the adversarial review pass A high-effort review (8 finder angles) of the change surfaced three confirmed defects, each fixed with a regression test that fails pre-fix: - **cancel() of a held turn stalled ~30s**: the session is already idle during the hold, so the interrupt may never emit the trailing idle the cancelled-settle path waited on. cancel() now settles a deferred turn immediately (its result is already in hand); this also prevents the force-cancel backstop from seeding a phantom orphan for a result that already passed. - **A followup's `is_error` / "Please run /login" could reject a held user turn** whose own result was a success (reachable with parallel subagents). Followup results now never touch the user-turn lifecycle. - **The followup's own trailing idle was un-owed**: settling at the followup result unblocks the client at exactly the moment that idle can lag past the next prompt's echo — which would have been read as the fresh turn ending without a result (#825 false-fail). All results now record their trailer debt. A second review round on the post-merge state (after resolving conflicts with #858) found and fixed five more: an echo-less command (`/context`) during a hold overwrote the held turn's outcome and hung (ensureActiveTurn now hands the held turn off and promotes the head); the refusal lane bypassed the deferral gate (result-time settles now route through a shared `settleOrDefer`); cancel()'s inline settle couldn't record the interrupt trailer's owed-idle debt (the counter moved onto the Session) and skipped the backstop disarm; `failAllTurns` rejected a held turn whose answer had already streamed (now resolves the stored outcome, like stream-done); and a followup result inside a cancel window skipped the trailer count. Settling at a followup result also now closes the #453 delivery stretch so a following replayed turn's fallback isn't suppressed. Known residuals, deliberately accepted: a subagent whose settle bookends are all lost parks the held turn until `session/cancel` or the next prompt (same rescue contract as the adapter's other wedge classes); a wake whose origin can't be attributed (the SDK's `origin` carries no task id) can settle the turn one followup early, degrading to today's behavior, never worse. ## Verification Unit: 530 tests green; the deferral/cancel/false-fail tests were each run against the pre-fix adapter and fail there. Live (real ACP stdio, CLI 2.1.206): held turn with in-turn subagent output + followup summary, resolution at the followup's result; plain prompts resolve instantly at their result. ## Out of scope (follow-ups) - A background **shell**'s exit waking the model between turns remains an out-of-turn episode — there is no turn to hold for it without blocking every prompt behind a dev server. Same for clients that want a session-level active/idle or background-task level signal (#864/#865's additive asks): both raw signals are already available today via `_meta.claudeCode.emitRawSDKMessages` → `_claude/sdkMessage` ext notifications. - Flipping the spawning `Task` tool call's status at terminal `task_updated` (#865's optional ask). A third review round on the consolidated state fixed: the reconciliation deleting live sync-subagent attribution entries (the level's universe is background-only — absent subagent entries are now kept but marked `endedPerLevel` so holds stop waiting on them); peer/coordinator/observer-origin results being treated as the user's turn (now gated via `AUTONOMOUS_RESULT_ORIGINS`, preserving channel-origin behavior); `emittedAssistantText` moving onto the Session so every held-turn settle lane closes the #453 delivery stretch (with the snapshot taken after the held-turn hand-off); the interrupt-trailer pre-count gated on a running cycle so idle-cancels can't mask future #825 detections; and `turnAwaitingSubagents` now reading `isSubagent` as defense in depth for the shells-never-defer contract. Plus `isHeldOpen()`/`disarmForceCancel()` deduplicating the hold predicate and backstop disarm. (An earlier follow-up — folding the attribution map and the live-subagent set into one task registry — is now done in this PR: `liveBackgroundTasks` keyed by task id, so the `background_tasks_changed` reconciliation heals both concerns and future terminal paths can't prune one but not the other.) Fixes #866. Addresses the settlement half of #864 / #865. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Fable 5 --- src/acp-agent.ts | 792 ++++++++++++++++++++----- src/tests/acp-agent.test.ts | 1088 ++++++++++++++++++++++++++++++++++- 2 files changed, 1709 insertions(+), 171 deletions(-) diff --git a/src/acp-agent.ts b/src/acp-agent.ts index 0c03fd5c..ea41692a 100644 --- a/src/acp-agent.ts +++ b/src/acp-agent.ts @@ -233,6 +233,53 @@ type Turn = { * never be drained by its result and would swallow an unrelated later * echo-less one instead. */ commandResultSeen?: boolean; + /** Task ids of the background subagents launched while this turn was the + * active one — including during its held-open drain window, so an agent + * chain (a followup that launches another subagent) extends the hold. + * A turn only waits on its OWN spawned subagents: a long-running agent + * from an earlier turn must not stall every later prompt's settlement. + * Known residual: task_started carries no lineage, so a spawn made by a + * PREVIOUS turn's followup chain while a later turn happens to be held + * is attributed to the holder — extending that hold behind a foreign + * chain. Bounded: the hold still ends at drain, hand-off, or cancel. */ + spawnedTaskIds?: Set; + /** Set instead of settling when the turn's terminal result arrives while + * subagents it spawned are still live (`spawnedTaskIds` ∩ + * `session.liveBackgroundTasks`). The turn is held open — its + * `session/prompt` stays pending — so the subagents' streamed output, + * their permission requests (which would otherwise block on an RPC a + * client that stops consuming at the prompt response never answers — + * issue #866), and the model's task-notification followup summary all + * land inside the turn. + * + * The CLI does NOT hold its trailing idle for background agents (observed + * on 2.1.206: `idle` follows the result immediately while the subagent + * still runs), so the hold spans multiple idle cycles: user result → + * idle → (subagent works) → task_notification → followup turn → idle. + * The stored outcome (the result's stop reason and usage snapshot) is + * what the turn settles with once its spawned subagents have settled — + * at the followup's terminal result (the summary has streamed by then), + * or at an idle with none of its subagents left (no followup came). A + * cancel or the next turn's echo hand-off settles it earlier, so a + * long-running subagent never holds the prompt hostage. + * + * Accepted residuals. (1) A subagent that ends WITHOUT waking the model — + * its task_notification lost or skipped (only the terminal task_updated + * patch is guaranteed per transition) — leaves no followup result and no + * further idle, so the held turn parks until `session/cancel` or the next + * prompt (either settles it: the echo hand-off or ensureActiveTurn's + * held-turn hand-off). Settling at the prune sites instead would preempt + * the followup summary in the normal ordering (prunes precede the + * notification), and a grace timer was judged not worth the machinery — + * the same rescue contract as the adapter's other wedge classes (issue + * #825's out-of-scope notes). (2) Drained-ness is judged by live-task + * membership only: with parallel subagents, a notification that prunes + * the last task during an earlier task's still-streaming followup lets + * that followup's result settle the turn before the LAST task's summary + * streams — degrading to post-turn delivery for it, never worse than the + * pre-hold behavior (pending wakes are not countable: notifications can + * batch into one followup). */ + deferredSettle?: PromptResponse; resolve: (response: PromptResponse) => void; reject: (error: unknown) => void; }; @@ -366,23 +413,116 @@ type Session = { * tool_use block streams; this set makes the two paths converge regardless of * order. Pruned at `tool_result` time alongside `toolUseCache`. */ emittedToolCalls: Set; - /** Maps a live subagent's agent id → the tool_use id of the Agent/Task call - * that spawned it. For subagent tasks the SDK keys its task registry by - * agent id, so `task_started.task_id` IS the `agentID` that `canUseTool` - * later receives, and `task_started.tool_use_id` is the spawning tool call. - * Populated from `task_started` and pruned when the task settles (a - * `task_notification` or a terminal `task_updated` patch). Lets the - * permission flow attribute a subagent's eagerly-emitted `tool_call` (and - * the permission request itself) to its parent tool call via - * `_meta.claudeCode.parentToolUseId`, matching the streamed subagent path. - * Best-effort: a `canUseTool` that races ahead of the consumer processing - * `task_started` omits the attribution from the eager tool_call, and the - * streamed tool_use chunk's refining `tool_call_update` — which carries the - * message-level `parent_tool_use_id` — restores it for merging clients; - * that recovery is what makes best-effort acceptable here. Non-subagent - * tasks (background Bash etc.) also pass through the map; see the - * `task_started` handler. */ - subagentParentToolUseIds: Map; + /** Registry of live background tasks, keyed by task id: populated at + * `task_started`, pruned when the task settles (a `task_notification` or + * a terminal `task_updated` patch), and reconciled against + * `background_tasks_changed`'s replace-semantics payload so a lost + * bookend can't leak an entry. One structure for both of its concerns so + * a future terminal path can't prune one and not the other: + * + * `parentToolUseId` — the tool_use id of the Agent/Task call that spawned + * the task. For subagent tasks the SDK keys its registry by agent id, so + * `task_started.task_id` IS the `agentID` that `canUseTool` later + * receives. Lets the permission flow attribute a subagent's + * eagerly-emitted `tool_call` (and the permission request itself) to its + * parent tool call via `_meta.claudeCode.parentToolUseId`, matching the + * streamed subagent path. Best-effort: a `canUseTool` that races ahead of + * the consumer processing `task_started` omits the attribution from the + * eager tool_call, and the streamed tool_use chunk's refining + * `tool_call_update` — which carries the message-level + * `parent_tool_use_id` — restores it for merging clients; that recovery + * is what makes best-effort acceptable here. + * + * `isSubagent` — whether the task is a Task/Agent-tool subagent + * (`task_started` carried a `subagent_type`). Read by + * `turnAwaitingSubagents` (with `spawnedTaskIds`) to decide whether a + * turn's settlement is deferred (see `Turn.deferredSettle`), so the + * subagents' post-result output and permission requests stay inside the + * turn (issues #864/#866). Deliberately false for non-subagent background + * tasks (e.g. a `run_in_background` dev server): those can outlive every + * turn, and the model's contract with them is a wake-on-exit + * notification, not a turn-scoped drain — a hold must NEVER wait on a + * shell. + * + * `endedPerLevel` — a `background_tasks_changed` payload did not include + * this subagent entry. The level's universe is BACKGROUND tasks only, so + * a live sync (foreground) subagent is legitimately absent — its entry is + * kept for permission attribution — but a hold must stop waiting on the + * id: an absent id can equally be a leaked async entry whose settle + * bookends were lost, and waiting on it would park the hold forever. + * Non-subagent entries are simply deleted instead (shells are always in + * the level's universe). */ + liveBackgroundTasks: Map< + string, + { + parentToolUseId?: string; + isSubagent: boolean; + /** Absent-from-level lifecycle, one field so the illegal + * armed-but-not-ended state is unrepresentable: undefined = live per + * the level signal; "ended" = a level omitted the task (holds stop + * waiting on it; attribution is kept); "sweep-armed" = a turn + * activation saw it ended — the NEXT activation deletes it. The + * one-activation grace exists for the absent-mark race (a level + * payload built before a live async agent's registration): a + * corrective inclusive level resets the field to undefined — one + * assignment, disarming any in-flight sweep — if it arrives within a + * full turn, keeping the agent's attribution; eager deletion would + * be irreversible, since levels never ADD entries. A re-mark + * preserves an in-flight arm (`??=`), keeping a continuously absent + * entry on its two-activation clock. */ + endedPerLevel?: "ended" | "sweep-armed"; + } + >; + /** Whether any top-level assistant text reached the client since the last + * stretch boundary. Set as a side effect of sending in the consumer's + * `sendUpdate`, never at an emission site; read at the terminal `result` + * to tell a turn whose answer was already delivered from one that only + * ever carried it on `result` (issue #453). Session-level (not + * consumer-scoped) so cancel()'s inline settle can clear it. + * + * The CURRENT boundary set — a new clear site must be added here: the + * result case's `finally` (user-turn results), settleActive's wasHeld + * clear (every held-turn settle lane: drain settle, both hand-offs, + * stream-done), failActive, the force-cancel backstop, the idle + * cancelled-settle, the autonomous-result close (only with no turn + * active OR queued — see its queued-turn guard), and cancel()'s inline + * mirror. + * + * Deliberately NOT reset on turn activation: activation can fire + * mid-message (see the echo hand-off), so a flag cleared there would + * forget text that already streamed and the result text would be emitted + * a second time. Neither the consolidated `assistant` message nor a + * `stream_event` carries `origin`, so an autonomous cycle's prose is + * indistinguishable from a user turn's here and sets the flag too; the + * autonomous-result close normally ends that stretch so a replayed + * prompt behind it still delivers, and only in the racing window (a + * turn already active or queued when the autonomous result lands) does + * the replayed turn stay silent rather than risk a duplicate. */ + emittedAssistantText: boolean; + /** The most recent `session_state_changed` state the consumer processed. + * Read by cancel() to decide whether the interrupt will produce a + * trailing idle worth pre-counting: interrupting a RUNNING cycle yields + * one; interrupting an already-idle session (the common held-turn shape) + * yields none, and a pre-counted debt that never drains would mask one + * future issue-#825 detection. */ + lastSessionState?: "idle" | "running" | "requires_action"; + /** How many trailing `session_state_changed: idle` messages are already + * accounted for: every result is followed by one (user-turn results that + * terminate a turn — settle, reject, or orphan skip — and autonomous + * cycles alike), as is a cancelled turn settled by the next turn's echo + * hand-off or by cancel()'s inline settle of a held turn whose interrupt + * pre-empts a running cycle — the reason this lives on the Session: + * cancel() must be able to record the debt. The idle handler absorbs + * owed idles; an idle that arrives when NONE is owed while the active + * turn is still unsettled means the SDK ended the turn without ever + * emitting its result, so the turn will never settle on its own (issue + * #825). Stream-level debt, deliberately NOT reset per turn: a lagged + * idle can arrive after the next turn has already activated (issue + * #773), and the debt is what attributes it to the turn that owed it. + * Over-counting (an idle the SDK never emits) is benign: the counter + * just absorbs one future idle, and detection degrades to the status quo + * rather than misfiring. */ + owedTrailingIdles: number; /** Maps the ACP `messageId` we expose to clients (see `messageIdForGrouping`) * to the SDK message uuid that the Agent SDK's rewind/resume APIs key on * (`Query.rewindFiles` takes a user-message uuid; `resumeSessionAt` takes an @@ -403,6 +543,52 @@ type Session = { messageIdToUuid: Map; }; +/** Result-message origin kinds that mark an AUTONOMOUS cycle — work the + * model did on its own (a task-notification followup, a peer/coordinator/ + * observer message it handled) rather than the user's prompt. Absent, + * `human`, and `channel` origins are the user's own turn (this adapter's + * prompts arrive as the ACP channel on some CLI configurations — ALL + * channel servers are treated as user, so a foreign channel integration's + * autonomously-handled result is misclassified as the user's; accepted, + * see below), and `auto-continuation` continues the user's turn, so its + * result is the turn's real terminal. + * + * Deliberately fail-OPEN: an unknown future kind defaults to the user + * lane. Misrouting a USER result into the autonomous lane hangs the + * prompt un-detectably (the result is skipped, its trailing idle absorbed + * as owed, so the #825 detector can't fire); misrouting an autonomous + * result into the user lane is the bounded misattribution class this set + * exists to reduce. */ +const AUTONOMOUS_RESULT_ORIGINS: ReadonlySet = new Set([ + "task-notification", + "peer", + "coordinator", + "observer", + "observer-activity", +]); + +/** Whether this turn's terminal result arrived but its settlement is being + * held for background subagents it spawned (see Turn.deferredSettle). The + * single spelling of the hold predicate, shared by the consumer's settle + * lanes and cancel(). */ +function isHeldOpen( + turn: Turn | null | undefined, +): turn is Turn & { deferredSettle: PromptResponse } { + return turn != null && turn.deferredSettle !== undefined && !turn.settled; +} + +/** Disarm the force-cancel backstop (see Session.forceCancelTimer). Every + * path that settles the active turn must run this so a timer can never fire + * on an already-settled turn — and must leave the field undefined, or the + * arm site's !forceCancelTimer guard would refuse to arm the backstop for + * the NEXT turn's cancel. */ +function disarmForceCancel(session: Session): void { + if (session.forceCancelTimer) { + clearTimeout(session.forceCancelTimer); + session.forceCancelTimer = undefined; + } +} + /** Compute a stable fingerprint of the session-defining params so we can * detect when a loadSession/resumeSession call requires tearing down and * recreating the underlying Query process. MCP servers are sorted by name @@ -1358,39 +1544,6 @@ export class ClaudeAcpAgent { // Stop reason accumulated for the active turn (result subtype, refusal, // max_tokens, …). Reset per turn; read when the turn settles at idle. let stopReason: StopReason = "end_turn"; - // Whether any top-level assistant text reached the client since the last - // stretch boundary — a user-turn result (see the result case's `finally`) - // or a turn torn down without one (failActive and the cancel settles). - // Set as a side effect of sending in `sendUpdate` below, never at an - // emission site. Read at the terminal `result` to tell a turn whose - // answer was already delivered from one that only ever carried it on - // `result` (issue #453). Deliberately NOT reset in resetTurnScratch: turn - // activation can fire mid-message (see there), so a flag cleared on - // activation would forget text that already streamed — the consolidated - // message then dedupes to nothing, nothing sets the flag again, and the - // result text would be emitted a second time. Neither the consolidated - // `assistant` message nor a `stream_event` carries `origin`, so a - // background followup's prose (or a compaction banner) is - // indistinguishable from a user turn's here and sets the flag too: a - // replayed turn right behind one stays silent rather than risk a - // duplicate, which is the pre-#453 behavior for that turn and never a - // double emission. - let emittedAssistantText = false; - // How many trailing `session_state_changed: idle` messages are already - // accounted for: every user-turn result that terminates a turn (settle, - // reject, or orphan skip) is followed by one, as is a cancelled turn - // settled by the next turn's echo hand-off. The idle handler absorbs owed - // idles; an idle that arrives when NONE is owed while the active turn is - // still unsettled means the SDK ended the turn without ever emitting its - // result, so the turn will never settle on its own (issue #825). - // Stream-level debt, deliberately NOT reset per turn: a lagged idle can - // arrive after the next turn has already activated (issue #773), and the - // debt is what attributes it to the turn that owed it. Over-counting (an - // idle the SDK never emits, e.g. CLI binaries without session-state - // events — issue #497) is benign: the counter just absorbs one future - // idle, and detection degrades to the status quo rather than misfiring. - let owedTrailingIdles = 0; - /** The consumer's single send chokepoint: every `sessionUpdate` in this * loop goes through here (never `this.client.sessionUpdate` directly) so * answer-delivery tracking is a property of sending, not something each @@ -1405,7 +1558,7 @@ export class ClaudeAcpAgent { const claudeMeta = update._meta?.claudeCode as { parentToolUseId?: string | null } | undefined; if (!claudeMeta?.parentToolUseId) { - emittedAssistantText = true; + session.emittedAssistantText = true; } } await this.client.sessionUpdate(notification); @@ -1448,6 +1601,29 @@ export class ClaudeAcpAgent { session.cancelled = false; session.pendingOrphanResults = 0; session.orphanCommands?.clear(); + // Two-phase sweep of registry entries the level signal ended (see + // the endedPerLevel field doc): armed at the first activation, + // deleted at the second — the same activation-time self-heal as the + // orphan lanes, and the growth bound for leaked entries whose settle + // bookends never arrive. The one-activation grace lets a corrective + // inclusive level rescue a live async agent that a racing payload + // absent-marked (deletion is irreversible: levels never ADD entries). + // Local-only commands don't advance the clock: two quick /context + // calls would otherwise burn the whole grace in seconds of wall time + // while the corrective level is still in flight, and they interact + // with no tasks — a later real turn still bounds growth. + if (!turn.isLocalOnlyCommand) { + for (const [taskId, record] of session.liveBackgroundTasks) { + if (!record.endedPerLevel) { + continue; + } + if (record.endedPerLevel === "sweep-armed") { + session.liveBackgroundTasks.delete(taskId); + } else { + record.endedPerLevel = "sweep-armed"; + } + } + } resetTurnScratch(); }; @@ -1468,7 +1644,27 @@ export class ClaudeAcpAgent { * result), so we skip those and only promote once the count is drained. */ const ensureActiveTurn = () => { if (session.activeTurn) { - return; + if (!isHeldOpen(session.activeTurn)) { + return; + } + // A held turn (Turn.deferredSettle) already produced its result, so + // this incoming user-turn result cannot be its — it belongs to the + // next queued command (an echo-less one, e.g. `/context` sent while + // the hold drains; a normal prompt's echo would have handed the held + // turn off before its result). Settle the held turn with its + // recorded outcome — the user moving on outranks the hold, same + // contract as the echo hand-off — and fall through to promote the + // queue head, which this result belongs to. Without this, the head + // would never be promoted (echo-less turns have no other promotion + // path) and its prompt would hang, while this result's outcome + // overwrote the held turn's. Orphan lanes below are necessarily + // empty while a turn is held: orphans are seeded by cancel(), which + // inline-settles a held turn, and activation cleared older ones. + // settleActive also closes the held turn's delivery stretch, so the + // promoted command's own delivery decision is not judged against the + // held turn's followup text (issue #453) — the caller snapshots the + // flag AFTER this runs. + settleActive(session.activeTurn.deferredSettle); } // Orphan accounting runs BEFORE the head check: an orphan's echo-less // result can arrive with an EMPTY queue (the common post-cancel @@ -1521,7 +1717,7 @@ export class ClaudeAcpAgent { return; } } - const head = (session.turnQueue ?? []).find((t) => !t.settled); + const head = firstUnsettledQueuedTurn(); if (!head) { return; } @@ -1559,6 +1755,62 @@ export class ClaudeAcpAgent { const findUnsettledTurn = (uuid: string) => (session.turnQueue ?? []).find((t) => t.promptUuid === uuid && !t.settled); + /** The first queued turn still awaiting its outcome, if any — the single + * spelling of "a prompt is pending" shared by the head promotion and + * the autonomous stretch-close guard. */ + const firstUnsettledQueuedTurn = () => (session.turnQueue ?? []).find((t) => !t.settled); + + /** Whether any background subagent this turn spawned is still live — + * while true, the turn's settlement stays deferred so the subagent's + * output and permission requests land inside it (see + * Turn.deferredSettle). */ + const turnAwaitingSubagents = (turn: Turn) => { + if (!turn.spawnedTaskIds?.size) { + return false; + } + for (const taskId of turn.spawnedTaskIds) { + const record = session.liveBackgroundTasks.get(taskId); + // The isSubagent read is defense in depth for the shells-never-defer + // contract: spawnedTaskIds only ever holds subagent ids today, but a + // future add site must not silently let a long-lived shell hold a + // prompt open. endedPerLevel entries are kept for attribution only — + // the level signal says the task is gone (or its bookends were + // lost), so a hold must not wait on them. + if (record?.isSubagent && !record.endedPerLevel) { + return true; + } + } + return false; + }; + + /** Settle the active turn's stored deferred outcome once none of its + * spawned subagents is live. The single drain rule shared by the + * followup-result and idle settle sites, so the two lanes can't drift. */ + const settleDeferredIfDrained = () => { + const turn = session.activeTurn; + if (isHeldOpen(turn) && !turnAwaitingSubagents(turn)) { + settleActive(turn.deferredSettle); + } + }; + + /** Settle the active turn with `outcome` now — unless subagents it + * spawned are still live, in which case store the outcome and hold the + * turn open (see Turn.deferredSettle). Every result-time settle of a + * turn that can have spawned subagents must route through here: a site + * calling settleActive directly bypasses the hold and re-opens the + * out-of-turn permission deadlock (issue #866) through its lane. */ + const settleOrDefer = (outcome: PromptResponse) => { + if ( + session.activeTurn && + !session.activeTurn.settled && + turnAwaitingSubagents(session.activeTurn) + ) { + session.activeTurn.deferredSettle = outcome; + } else { + settleActive(outcome); + } + }; + /** Settle the active turn's deferred exactly once, disarm the force-cancel * backstop (the turn is over), and drop it from the queue. */ const settleActive = (result: PromptResponse) => { @@ -1566,24 +1818,35 @@ export class ClaudeAcpAgent { if (!turn || turn.settled) { return; } + // Captured before the settled flip below (isHeldOpen tests !settled). + const wasHeld = isHeldOpen(turn); turn.settled = true; - if (session.forceCancelTimer) { - clearTimeout(session.forceCancelTimer); - session.forceCancelTimer = undefined; - } + disarmForceCancel(session); session.turnQueue = (session.turnQueue ?? []).filter((t) => t !== turn); session.activeTurn = null; streamedToolInputs.clear(); + if (wasHeld) { + // Settling a held turn is its delivery-stretch boundary: the turn's + // answer finished long ago, so text streamed since the last boundary + // is normally its followups' — left latched it would suppress a + // following replayed turn's issue-#453 result-text fallback (the + // common post-hold sequence). Known trade: at the echo hand-off an + // incoming turn's pre-echo deltas share this one boolean, so a + // STREAMING replay on a usage-omitting backend could re-emit its + // answer — the flag cannot attribute text to a turn before its + // echo, and the suppression direction is the common one, so the + // clear wins. Every held-settle lane inherits this: the drain + // settle, both hand-offs, and stream-done; cancel()'s inline mirror + // carries its own copy. + session.emittedAssistantText = false; + } turn.resolve(result); }; /** Reject the active turn (auth required, error result, …) without tearing * down the consumer: the stream continues to idle and later turns proceed. */ const failActive = (error: unknown) => { - if (session.forceCancelTimer) { - clearTimeout(session.forceCancelTimer); - session.forceCancelTimer = undefined; - } + disarmForceCancel(session); const turn = session.activeTurn; if (!turn || turn.settled) { return; @@ -1596,16 +1859,13 @@ export class ClaudeAcpAgent { // #825 idle-fail) never see the result whose `finally` would close it — // start the next stretch clean, or its stale delivery record would // suppress the next turn's issue-#453 result-text fallback. - emittedAssistantText = false; + session.emittedAssistantText = false; turn.reject(error); }; /** Reject every in-flight turn — used when the stream dies. */ const failAllTurns = (error: unknown) => { - if (session.forceCancelTimer) { - clearTimeout(session.forceCancelTimer); - session.forceCancelTimer = undefined; - } + disarmForceCancel(session); const turns = session.activeTurn ? [session.activeTurn, ...(session.turnQueue ?? []).filter((t) => t !== session.activeTurn)] : [...(session.turnQueue ?? [])]; @@ -1613,8 +1873,17 @@ export class ClaudeAcpAgent { session.turnQueue = []; for (const turn of turns) { if (!turn.settled) { + const wasHeld = isHeldOpen(turn); turn.settled = true; - turn.reject(error); + if (wasHeld) { + // A held turn's answer already streamed and its outcome is + // recorded — a stream death during the post-answer hold is a + // background failure, not the turn's. Resolve with the real + // outcome, mirroring the stream-done path. + turn.resolve(turn.deferredSettle); + } else { + turn.reject(error); + } } } }; @@ -1695,7 +1964,7 @@ export class ClaudeAcpAgent { // streamed text can't suppress the next turn's issue-#453 fallback. // If a late orphan result does arrive, its `finally` clears again; // FIFO ordering means no live turn's text can have streamed yet. - emittedAssistantText = false; + session.emittedAssistantText = false; // If the session is being torn down, abandon the in-flight next() // (swallowing any later rejection so it can't surface as unhandled) // and stop; otherwise re-arm and keep consuming — `pendingNext` @@ -1722,11 +1991,17 @@ export class ClaudeAcpAgent { // turn's real outcome. // // Settle the turn that was in flight so its prompt() doesn't hang: - // cancelled if a cancel is pending, otherwise the accumulated outcome. - settleActive({ - stopReason: session.cancelled ? "cancelled" : stopReason, - usage: sessionUsage(session), - }); + // cancelled if a cancel is pending, otherwise the outcome a + // deferred turn already recorded (see Turn.deferredSettle) or the + // accumulated scratch outcome. The scratch currently still equals + // a deferred turn's stored outcome (followup results never mutate + // it), but the stored one is the authoritative source. + const inFlight = session.activeTurn; + settleActive( + session.cancelled + ? { stopReason: "cancelled", usage: sessionUsage(session) } + : (inFlight?.deferredSettle ?? { stopReason, usage: sessionUsage(session) }), + ); // Queued turns the SDK never started never ran, so reject them rather // than reporting a success (end_turn) — or a misleading "cancelled" — // for a prompt that produced no output. (A cancel already settled the @@ -1947,6 +2222,7 @@ export class ClaudeAcpAgent { break; } case "session_state_changed": { + session.lastSessionState = message.state; if (message.state === "idle") { // A non-cancelled turn normally settled at its terminal // `result` already (issue #773), and that result recorded an @@ -1984,14 +2260,31 @@ export class ClaudeAcpAgent { // delivery stretch here: idle is the SDK's authoritative // turn-over signal, and stale partial-text state would // suppress the next turn's issue-#453 fallback. - emittedAssistantText = false; - } else if (owedTrailingIdles > 0) { + session.emittedAssistantText = false; + } else if (isHeldOpen(session.activeTurn)) { + // A turn held open for its background subagents (see + // Turn.deferredSettle). Idles keep their normal cadence + // during the hold — the CLI emits one per processing + // cycle (the turn's own trailer, then one per followup), + // NOT one final "all drained" signal — so each one + // absorbs an outstanding trailer debt, and the turn only + // settles once none of its spawned subagents is left + // (the followup-result settle usually got there first; + // this is the fallback when no followup came). Mid-hold + // idles never fall through: a held turn HAS its result, + // so reading its idle as "turn abandoned without a + // result" (issue #825) would fail a healthy prompt. + if (session.owedTrailingIdles > 0) { + session.owedTrailingIdles--; + } + settleDeferredIfDrained(); + } else if (session.owedTrailingIdles > 0) { // Absorb a settled turn's trailing idle. Also covers a // cancel that landed between a turn's counted result and // this lagged idle (no active turn to settle): the idle // still belongs to that settled turn, and skipping the // decrement would leak the debt permanently. - owedTrailingIdles--; + session.owedTrailingIdles--; } else if ( !session.cancelled && session.activeTurn && @@ -2155,32 +2448,47 @@ export class ClaudeAcpAgent { case "task_started": // For subagent tasks `task_id` is the subagent's agent id (the // SDK keys its task registry by agent id) and `tool_use_id` is - // the Agent/Task tool_use that spawned it. Record the mapping so - // the subagent's permission requests — which reach canUseTool - // with only `agentID` — can attribute their eagerly-emitted + // the Agent/Task tool_use that spawned it — recorded so the + // subagent's permission requests, which reach canUseTool with + // only `agentID`, can attribute their eagerly-emitted // tool_call to the parent tool call. Non-subagent tasks (e.g. - // background Bash) land here too; their task_ids never match an - // agentID, so the stray entries are inert until pruned below. - if (message.tool_use_id) { - session.subagentParentToolUseIds.set(message.task_id, message.tool_use_id); + // background Bash) land here too; their task_ids never match + // an agentID, so those entries are inert for attribution. + // + // `isSubagent` marks Task/Agent-tool subagents — the tasks + // whose completion wakes the model for a followup, so the + // ones worth deferring turn settlement for. A sync subagent + // is pruned (terminal task_updated) before its turn's result + // can arrive, so registry membership at result time means an + // async subagent. Their spawn is also recorded on the active + // turn: a turn only ever waits on its own subagents, and a + // spawn during a held-open drain window (an agent chain) + // extends that turn's hold. + session.liveBackgroundTasks.set(message.task_id, { + parentToolUseId: message.tool_use_id, + isSubagent: !!message.subagent_type, + }); + if (message.subagent_type && session.activeTurn && !session.activeTurn.settled) { + (session.activeTurn.spawnedTaskIds ??= new Set()).add(message.task_id); } break; case "task_notification": - // The task settled — no further tool calls can originate from - // it, so the attribution mapping can be dropped. - session.subagentParentToolUseIds.delete(message.task_id); + // The task settled — no further tool calls can originate + // from it, so its registry entry can be dropped. + session.liveBackgroundTasks.delete(message.task_id); break; case "task_updated": // terminal-status task_updated patch and a (deduplicated) // task_notification when a task settles, but only the patch is - // guaranteed per transition — prune on it too so the map can't - // grow for the session's lifetime if a notification is skipped. + // guaranteed per transition — prune on it too so the registry + // can't grow for the session's lifetime if a notification is + // skipped. if ( message.patch.status === "completed" || message.patch.status === "failed" || message.patch.status === "killed" ) { - session.subagentParentToolUseIds.delete(message.task_id); + session.liveBackgroundTasks.delete(message.task_id); } break; case "worker_shutting_down": @@ -2272,12 +2580,49 @@ export class ClaudeAcpAgent { break; // `control_request_progress` only reports on side_question // control requests, which this adapter never issues. - // `background_tasks_changed` is a level signal (the full live - // background-task set on every membership change) for surfaces - // that render a background-activity indicator; turn lifecycle - // here is driven by results/idle, so there is nothing to track. case "control_request_progress": + break; case "background_tasks_changed": + // A level signal: the full live background-task set on every + // membership change, with REPLACE semantics. Used only to + // reconcile `liveBackgroundTasks` — dropping (or, for + // subagent entries, unpinning) any entry whose settle + // bookend (task_notification / terminal task_updated) was + // lost, so a leaked subagent entry can't defer its spawning + // turn's settlement forever. Growth of retained + // (endedPerLevel) subagent entries is bounded by the + // activation-time sweep in activateTurn, not here. It never + // ADDS entries (the payload carries no attribution or + // subagent marker), so the unspecified ordering vs. the edge + // bookends is safe: a level that precedes its task_started + // simply no-ops here. + if (session.liveBackgroundTasks.size > 0) { + const live = new Set(message.tasks.map((t) => t.task_id)); + for (const [taskId, record] of session.liveBackgroundTasks) { + if (live.has(taskId)) { + // The level proves the task live in the background + // universe (e.g. a foreground agent was backgrounded + // after an earlier absent-marking, or that marking was + // a racing payload built before the task registered) — + // un-end it so a hold waits on it again, and disarm + // the activation sweep. + record.endedPerLevel = undefined; + continue; + } + if (record.isSubagent) { + // The level's universe is BACKGROUND tasks only, so a + // live sync (foreground) subagent is legitimately + // absent — deleting its entry would strand its + // permission attribution (#859). Keep the entry but + // stop any hold from waiting on the id: an absent id + // can equally be a leaked async entry whose settle + // bookends were lost. + record.endedPerLevel ??= "ended"; + } else { + session.liveBackgroundTasks.delete(taskId); + } + } + } break; default: unreachable(message, this.logger); @@ -2285,28 +2630,21 @@ export class ClaudeAcpAgent { } break; case "result": { - // Task-notification followups are autonomous work triggered by a - // task-notification system message, not by the user's prompt. - // They should not influence the user-turn lifecycle (stop reason, - // slash-command output forwarding) but their cost is real. - const isTaskNotification = message.origin?.kind === "task-notification"; + // A result from an autonomous cycle — a task-notification + // followup, or a peer/coordinator/observer message the model + // handled on its own (see AUTONOMOUS_RESULT_ORIGINS) — is not + // the user's prompt's. Autonomous results must never touch the + // user-turn lifecycle (stop reason, settles, failActive, + // slash-command output forwarding), though their cost is real. + const isAutonomousResult = + message.origin != null && AUTONOMOUS_RESULT_ORIGINS.has(message.origin.kind); - // A result closes the stretch of output it terminates: snapshot - // the delivery record before the handling below can emit anything - // of its own, and clear it in the `finally` so every exit from - // this case — the cancelled-guard and refusal breaks included — - // starts the next stretch clean. Clearing up front instead would - // let result-time emissions (refusal explanation, result-text - // forwarding) taint the next stretch and suppress a following - // replayed turn's fallback. Task-notification followups run - // alongside a user turn and must not clear its flag. - const deliveredAssistantText = emittedAssistantText; try { // Reconcile the Fast mode toggle with the SDK's reported state. - // Gated to user-driven turns like every other side effect below; a - // background followup's state lands on the next user turn's result. - // Runs even when the turn errors or was cancelled. - if (!isTaskNotification) { + // Gated to user-driven turns like every other side effect below; + // an autonomous cycle's state lands on the next user turn's + // result. Runs even when the turn errors or was cancelled. + if (!isAutonomousResult) { await this.syncFastModeState(params.sessionId, session, message.fast_mode_state); } @@ -2319,11 +2657,27 @@ export class ClaudeAcpAgent { // commands whose shared or late result this is, even when the // result is the ACTIVE turn's (ensureActiveTurn never looks at // the map in that case). - if (!isTaskNotification) { + if (!isAutonomousResult) { recordResultForOrphanCommands(); ensureActiveTurn(); } + // A result closes the stretch of output it terminates: snapshot + // the delivery record — AFTER ensureActiveTurn, whose held-turn + // hand-off closes the held stretch, so an echo-less command + // promoted here is judged on its own delivery, not on the held + // turn's followup text — and before the handling below can emit + // anything of its own; the `finally` then clears it so every + // exit from this case (the cancelled-guard and refusal breaks + // included) starts the next stretch clean. Clearing up front + // instead would let result-time emissions (refusal explanation, + // result-text forwarding) taint the next stretch and suppress a + // following replayed turn's fallback. Autonomous cycles run + // alongside a user turn and must not clear its flag (they exit + // through the early break below, which the gated `finally` + // leaves alone). + const deliveredAssistantText = session.emittedAssistantText; + // Every user-turn result terminates a turn (settle, reject, or // orphan skip) and the SDK follows it with a trailing // `session_state_changed: idle` — record the debt so the idle @@ -2338,17 +2692,30 @@ export class ClaudeAcpAgent { // late result after the backstop settled it — get no such settle, // so their trailers must be counted here or they'd later be read // as the next healthy turn being abandoned and false-fail it. - if (!isTaskNotification && (!session.cancelled || !session.activeTurn)) { - owedTrailingIdles++; + // Autonomous results (followups, peer/channel/coordinator + // cycles) are counted too: each is its own processing cycle with + // its own trailing idle, and that idle can lag past the next + // prompt's echo — which, un-owed, would be read as the fresh + // turn being abandoned (#825 false-fail). That lag was mostly + // unreachable when such cycles only ran with no pending turn, + // but a held turn settling AT a followup result unblocks the + // client at exactly that point, making the race the common + // case. + // The cancelled-ACTIVE-turn exclusion applies only to that + // turn's OWN result — a followup result arriving inside the + // cancel window still gets its own trailer and must be counted, + // or that idle would later false-fail the next prompt. + if (isAutonomousResult || !session.cancelled || !session.activeTurn) { + session.owedTrailingIdles++; } - // Accumulate usage into the user turn's tally. Skip task-notification - // followups: their cost is real but is reported separately via the + // Accumulate usage into the user turn's tally. Skip autonomous + // results: their cost is real but is reported separately via the // usage_update below, and `session.accumulatedUsage` is only reset on - // turn activation — so folding a task-notification result that lands + // turn activation — so folding an autonomous result that lands // after the next turn is active (but before it settles) would leak // those tokens into that turn's PromptResponse.usage. - if (!isTaskNotification) { + if (!isAutonomousResult) { session.accumulatedUsage.inputTokens += message.usage.input_tokens; session.accumulatedUsage.outputTokens += message.usage.output_tokens; session.accumulatedUsage.cachedReadTokens += message.usage.cache_read_input_tokens; @@ -2387,19 +2754,51 @@ export class ClaudeAcpAgent { } if (session.cancelled) { - if (!isTaskNotification) { + if (!isAutonomousResult) { stopReason = "cancelled"; } break; } + // A held turn (see Turn.deferredSettle) settles at its + // followup's terminal result: this is the earliest point at which + // the promised summary has fully streamed — the trailing idle + // would work too, but a client should not wait out another idle + // round-trip for a response whose content is already complete. + // (While the turn still awaits another of its subagents — + // parallel spawns — the helper holds; the next notification's + // followup settles it instead. Other autonomous origins — peer/ + // channel/coordinator cycles — reach here too: settling a + // drained hold at their results is as good as the idle + // fallback.) Then stop: everything below is user-turn + // lifecycle, and an autonomous outcome must never touch it — + // its is_error or "Please run /login" text would otherwise + // failActive a live turn (the held one, or the user's next + // prompt) whose own result recorded a different outcome. + if (isAutonomousResult) { + settleDeferredIfDrained(); + // With no turn in flight OR QUEUED (also after the settle + // above), the stretch holds only autonomous prose — close + // it, so a replayed next prompt isn't silently suppressed by + // the issue-#453 delivery check. A live turn's flag may + // guard the USER's already-streamed text — and so may a + // QUEUED turn's: with mid-message echo lag its deltas stream + // before the echo activates it (activeTurn still null), and + // clearing then would re-emit that answer via the fallback, + // the duplicate direction the flag's doc forbids. + if (!session.activeTurn && !firstUnsettledQueuedTurn()) { + session.emittedAssistantText = false; + } + break; + } + // A refusal can arrive on any result subtype (and may even set // is_error), so handle it before the subtype switch — otherwise the // is_error throw below would surface it as an internal error. The // refused assistant message carries no visible content, so surface // the classifier's explanation (when available) and report ACP's // dedicated `refusal` stop reason. - if (message.stop_reason === "refusal" && !isTaskNotification) { + if (message.stop_reason === "refusal") { if (lastRefusalExplanation) { await sendUpdate({ sessionId: params.sessionId, @@ -2410,7 +2809,12 @@ export class ClaudeAcpAgent { }); } stopReason = "refusal"; - settleActive({ stopReason: "refusal", usage: sessionUsage(session) }); + // Through the deferral gate, not settleActive: a refusal can + // land on a turn whose spawned subagents are still live, and + // settling it out from under them would strand their output + // and permission requests out-of-turn (issue #866's deadlock, + // through the refusal lane). + settleOrDefer({ stopReason: "refusal", usage: sessionUsage(session) }); break; } @@ -2421,9 +2825,7 @@ export class ClaudeAcpAgent { break; } if (message.stop_reason === "max_tokens") { - if (!isTaskNotification) { - stopReason = "max_tokens"; - } + stopReason = "max_tokens"; break; } if (message.is_error) { @@ -2448,13 +2850,12 @@ export class ClaudeAcpAgent { // token fields (see snapshotFromUsage), and the replay lane // was reported from exactly such a backend — treat a missing // count as the replay signature rather than silently disabling - // the fallback there. Task-notification followups never - // originate from a user slash command and must not inject - // background prose into the feed — always skip. + // the fallback there. (Autonomous results never get here — + // they exit at the early break above — so no background + // prose can be injected into the feed.) if ( - !isTaskNotification && - (session.activeTurn?.isLocalOnlyCommand || - (!deliveredAssistantText && (message.usage.output_tokens ?? 0) === 0)) + session.activeTurn?.isLocalOnlyCommand || + (!deliveredAssistantText && (message.usage.output_tokens ?? 0) === 0) ) { for (const notification of toAcpNotifications( message.result, @@ -2471,9 +2872,7 @@ export class ClaudeAcpAgent { } case "error_during_execution": { if (message.stop_reason === "max_tokens") { - if (!isTaskNotification) { - stopReason = "max_tokens"; - } + stopReason = "max_tokens"; break; } if (message.is_error) { @@ -2485,9 +2884,7 @@ export class ClaudeAcpAgent { ); break; } - if (!isTaskNotification) { - stopReason = "end_turn"; - } + stopReason = "end_turn"; break; } case "error_max_budget_usd": @@ -2502,9 +2899,7 @@ export class ClaudeAcpAgent { ); break; } - if (!isTaskNotification) { - stopReason = "max_turn_requests"; - } + stopReason = "max_turn_requests"; break; default: unreachable(message, this.logger); @@ -2514,15 +2909,33 @@ export class ClaudeAcpAgent { // as soon as the answer is done, rather than waiting for the SDK's // trailing `idle` (which can lag while background work runs — issue // #773). The consumer keeps draining afterward (absorbing idle and - // forwarding any background output). is_error/auth already settled - // via failActive; cancellation is left to the idle/abort path. - // settleActive is idempotent, so a duplicate idle is a no-op. - if (!isTaskNotification && !session.cancelled) { - settleActive({ stopReason, usage: sessionUsage(session) }); + // forwarding any background output). + // + // One exception: while background subagents this turn spawned are + // still live, settling now would strand their remaining work + // outside any turn — ACP allows out-of-turn session/update, but + // many clients stop consuming at the prompt response, and a + // subagent's permission request would block on an RPC nobody + // answers (issues #864/#866). Hold the turn open instead: store + // the outcome and settle with it once the subagents are done — + // at their followup's terminal result (see the deferred-settle + // block above the subtype switch) or at an idle with none of + // them left — so the subagents' streamed output, their + // permission requests, and the model's promised summary all land + // inside the turn. `session/cancel` and the next prompt's echo + // hand-off still settle a deferred turn early, so a long-running + // subagent never holds the prompt hostage. + // + // is_error/auth already settled via failActive (activeTurn is null + // then, so both branches no-op); cancellation is left to the + // idle/abort path. settleActive is idempotent, so a duplicate + // idle is a no-op. + if (!session.cancelled) { + settleOrDefer({ stopReason, usage: sessionUsage(session) }); } } finally { - if (!isTaskNotification) { - emittedAssistantText = false; + if (!isAutonomousResult) { + session.emittedAssistantText = false; } } break; @@ -2688,16 +3101,33 @@ export class ClaudeAcpAgent { // debt so that lagged idle is absorbed rather than read // as the freshly-activated turn ending without a result // (which would false-fail a healthy turn — issue #825). - owedTrailingIdles++; + // Counted for a DEFERRED turn too, even though its own + // result already recorded a debt that may still be + // outstanding: the interrupt can produce a trailer of + // its own, and over-counting is benign (absorbs one + // future idle) while under-counting risks the false + // fail this debt exists to prevent. + session.owedTrailingIdles++; // Before activateTurn resets the accumulator, so the // usage still belongs to the cancelled turn. settleActive({ stopReason: "cancelled", usage: sessionUsage(session) }); + } else if (isHeldOpen(session.activeTurn)) { + // A turn held open for its background subagents (see + // Turn.deferredSettle) hands off with the real outcome + // its result recorded, not a guessed end_turn — the + // user moving on must not block behind a long-running + // subagent, but it must not rewrite the stop reason + // either. Its trailing-idle debt stands and is absorbed + // when the drain idle eventually arrives. + settleActive(session.activeTurn.deferredSettle); } else { settleActive({ stopReason: "end_turn", usage: sessionUsage(session) }); } } // Unlike the no-result teardown lanes, this hand-off must - // NOT clear emittedAssistantText: the echo can land + // NOT clear emittedAssistantText for a NON-held previous + // turn (a held one's settleActive above closes its own + // stretch): the echo can land // mid-message, so deltas already streamed belong to the turn // being activated — clearing would forget them and let its // result re-emit the answer. @@ -3091,6 +3521,55 @@ export class ClaudeAcpAgent { ); } + // A deferred active turn (see Turn.deferredSettle) already has its + // result — it is only held open for its background subagents, which the + // interrupt below tears down. Settle it "cancelled" NOW: during the hold + // the session is typically already in state idle (the CLI's trailer + // fired at the result), so the interrupt may produce no fresh idle for + // the consumer's cancelled-settle path to run on, and the cancel would + // otherwise stall until the force-cancel backstop. Any outstanding + // trailer debt is absorbed by the idle handler when its idle does come. + // The turn's own usage snapshot is reported per the cancelled-usage + // contract (issue #844). + { + const active = session.activeTurn; + if (isHeldOpen(active)) { + active.settled = true; + // Mirror settleActive's invariants (it is consumer-scoped and + // unreachable from here): disarm the backstop — none should be + // armed for a held turn, but a drift here must not leave a timer + // firing on a settled turn — and drop the turn from the queue. + disarmForceCancel(session); + session.turnQueue = (session.turnQueue ?? []).filter((t) => t !== active); + session.activeTurn = null; + // Settling a held turn closes its delivery stretch: any streamed + // text since the last boundary was its followups', and left latched + // it would suppress a following replayed turn's issue-#453 fallback. + session.emittedAssistantText = false; + // When the interrupt below pre-empts a live cycle — running, or + // blocked on a permission request (requires_action, the #866 shape + // users cancel out of) — it produces a trailer idle with no counted + // result; with the hold's own trailer typically already absorbed, + // that idle would be un-owed and could lag past the next prompt's + // echo — read as the fresh turn ending without a result (issue #825 + // false-fail). Pre-count it unless the session sits idle: there the + // interrupt emits nothing, and a debt that never drains would mask + // one future #825 detection. (lastSessionState is last-CONSUMED, so + // both stale reads exist and both are accepted one-cycle windows: a + // running transition still in the backlog reads as stale idle and + // under-counts — that false-fail additionally needs the trailer to + // lag past the next echo — while a cycle already completed into the + // backlog reads as stale non-idle and over-counts, masking one + // future #825 detection. Undefined — no state event consumed — + // pre-counts; that only occurs on CLIs whose missing idle events + // also disable the detector the debt could mask.) + if (session.lastSessionState !== "idle") { + session.owedTrailingIdles++; + } + active.resolve({ stopReason: "cancelled", usage: active.deferredSettle.usage }); + } + } + // Arm a backstop before interrupting: if a turn is actively consuming the // query and interrupt() doesn't make the SDK yield (e.g. a wedged TaskOutput // block — issue #680), force the consumer to settle the active turn @@ -3207,10 +3686,7 @@ export class ClaudeAcpAgent { // after the floor, and clear the timer so it can't outlive the deleted // session (it isn't unref'd and would otherwise keep the event loop alive // until it fires). - if (session.forceCancelTimer) { - clearTimeout(session.forceCancelTimer); - session.forceCancelTimer = undefined; - } + disarmForceCancel(session); session.cancelController?.abort(); this.closeQueryStream(session); // Abort the SDK abort signal only on explicit destroy. closeQueryStream @@ -3558,8 +4034,10 @@ export class ClaudeAcpAgent { // When the tool call originates inside a subagent, attribute the eagerly // emitted tool_call (and the permission request itself) to the Agent/Task // tool call that spawned the subagent, mirroring the streamed subagent - // path's `_meta.claudeCode.parentToolUseId` (see `subagentParentToolUseIds`). - const parentToolUseId = agentID ? session.subagentParentToolUseIds.get(agentID) : undefined; + // path's `_meta.claudeCode.parentToolUseId` (see `liveBackgroundTasks`). + const parentToolUseId = agentID + ? session.liveBackgroundTasks.get(agentID)?.parentToolUseId + : undefined; if (agentID && !parentToolUseId) { // The attribution rests on an undocumented SDK invariant // (task_started.task_id === canUseTool's agentID for subagent tasks; @@ -4712,7 +5190,9 @@ export class ClaudeAcpAgent { taskState, toolUseCache: {}, emittedToolCalls: new Set(), - subagentParentToolUseIds: new Map(), + liveBackgroundTasks: new Map(), + emittedAssistantText: false, + owedTrailingIdles: 0, messageIdToUuid: new Map(), }; diff --git a/src/tests/acp-agent.test.ts b/src/tests/acp-agent.test.ts index 9d6725a5..8ac9fdcb 100644 --- a/src/tests/acp-agent.test.ts +++ b/src/tests/acp-agent.test.ts @@ -160,7 +160,9 @@ function mockSessionState(overrides: Record = {}) { taskState: new Map(), toolUseCache: {}, emittedToolCalls: new Set(), - subagentParentToolUseIds: new Map(), + liveBackgroundTasks: new Map(), + emittedAssistantText: false, + owedTrailingIdles: 0, messageIdToUuid: new Map(), ...overrides, } as any; @@ -1948,7 +1950,9 @@ describe("permission request cancellation", () => { taskState: new Map(), toolUseCache: {}, emittedToolCalls: new Set(), - subagentParentToolUseIds: new Map(), + liveBackgroundTasks: new Map(), + emittedAssistantText: false, + owedTrailingIdles: 0, messageIdToUuid: new Map(), } as any; return agent.sessions[sessionId]!; @@ -2323,7 +2327,10 @@ describe("subagent permission attribution (issue #851)", () => { it("attributes a subagent tool's eager tool_call and permission request to the spawning tool call", async () => { const { agent, updates, requests, session } = setup(); - session.subagentParentToolUseIds.set("agent-42", "toolu_parent"); + session.liveBackgroundTasks.set("agent-42", { + parentToolUseId: "toolu_parent", + isSubagent: true, + }); await agent.canUseTool("session-1")("Bash", { command: "ls" }, { signal: new AbortController().signal, @@ -2455,9 +2462,9 @@ describe("subagent permission attribution (issue #851)", () => { await agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "go" }] }); - expect(agent.sessions["test-session"]!.subagentParentToolUseIds.get("agent-42")).toBe( - "toolu_parent", - ); + expect( + agent.sessions["test-session"]!.liveBackgroundTasks.get("agent-42")?.parentToolUseId, + ).toBe("toolu_parent"); }); it("prunes the mapping when the task settles (task_notification)", async () => { @@ -2489,7 +2496,7 @@ describe("subagent permission attribution (issue #851)", () => { await agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "go" }] }); - expect(agent.sessions["test-session"]!.subagentParentToolUseIds.size).toBe(0); + expect(agent.sessions["test-session"]!.liveBackgroundTasks.size).toBe(0); }); it("prunes the mapping on a terminal task_updated patch (belt and braces)", async () => { @@ -2529,9 +2536,9 @@ describe("subagent permission attribution (issue #851)", () => { await agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "go" }] }); - const map = agent.sessions["test-session"]!.subagentParentToolUseIds; + const map = agent.sessions["test-session"]!.liveBackgroundTasks; expect(map.has("agent-42")).toBe(false); - expect(map.get("agent-43")).toBe("toolu_parent_2"); + expect(map.get("agent-43")?.parentToolUseId).toBe("toolu_parent_2"); }); }); @@ -3726,7 +3733,9 @@ describe("session/close", () => { taskState: new Map(), toolUseCache: {}, emittedToolCalls: new Set(), - subagentParentToolUseIds: new Map(), + liveBackgroundTasks: new Map(), + emittedAssistantText: false, + owedTrailingIdles: 0, messageIdToUuid: new Map(), }; return agent.sessions[sessionId]!; @@ -3814,7 +3823,9 @@ describe("session/delete", () => { taskState: new Map(), toolUseCache: {}, emittedToolCalls: new Set(), - subagentParentToolUseIds: new Map(), + liveBackgroundTasks: new Map(), + emittedAssistantText: false, + owedTrailingIdles: 0, messageIdToUuid: new Map(), }; return agent.sessions[sessionId]!; @@ -3919,7 +3930,9 @@ describe("getOrCreateSession param change detection", () => { taskState: new Map(), toolUseCache: {}, emittedToolCalls: new Set(), - subagentParentToolUseIds: new Map(), + liveBackgroundTasks: new Map(), + emittedAssistantText: false, + owedTrailingIdles: 0, messageIdToUuid: new Map(), }; return agent.sessions[sessionId]!; @@ -6386,7 +6399,9 @@ describe("post-error recovery", () => { taskState: new Map(), toolUseCache: {}, emittedToolCalls: new Set(), - subagentParentToolUseIds: new Map(), + liveBackgroundTasks: new Map(), + emittedAssistantText: false, + owedTrailingIdles: 0, messageIdToUuid: new Map(), }; return { interrupt }; @@ -7982,6 +7997,1045 @@ describe("post-error recovery", () => { }); }); +describe("deferred settlement for live background subagents (issues #864/#866)", () => { + // A turn whose terminal result arrives while background subagents IT + // spawned are still live must NOT settle at that result: ACP allows + // out-of-turn session/update, but many clients stop consuming at the + // prompt response, so the subagents' remaining output would be dropped + // and their permission requests would block on an RPC nobody answers. The turn is held open + // across the CLI's idle cycles (observed cadence: user result → idle → + // subagent works → task_notification → followup turn → idle) and settles + // once its subagents are done — at the followup's terminal result, or at + // an idle with none of them left. + + function createMockAgent() { + const mockClient = { + sessionUpdate: async () => {}, + } as unknown as AcpClient; + return new ClaudeAcpAgent(mockClient, { log: () => {}, error: () => {} }); + } + + const waitFor = async (cond: () => boolean) => { + for (let i = 0; i < 200; i++) { + if (cond()) return; + await new Promise((r) => setTimeout(r, 0)); + } + throw new Error("waitFor timed out"); + }; + + function resultMessage(overrides: Record = {}) { + return { + type: "result" as const, + subtype: "success" as const, + stop_reason: "end_turn", + is_error: false, + result: "", + errors: [], + duration_ms: 0, + duration_api_ms: 0, + num_turns: 1, + total_cost_usd: 0, + usage: { + input_tokens: 10, + output_tokens: 5, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + }, + modelUsage: {}, + permission_denials: [], + uuid: randomUUID(), + session_id: "test-session", + ...overrides, + }; + } + + const stateChanged = (state: "running" | "idle" | "requires_action") => ({ + type: "system", + subtype: "session_state_changed", + state, + uuid: randomUUID(), + session_id: "test-session", + }); + const running = () => stateChanged("running"); + const idle = () => stateChanged("idle"); + const requiresAction = () => stateChanged("requires_action"); + + /** Agent whose client records top-level agent_message_chunk texts as + * `chunk:` in the returned events array. */ + const chunkCapturingAgent = () => { + const events: string[] = []; + const mockClient = { + sessionUpdate: async (u: any) => { + if (u.update?.sessionUpdate === "agent_message_chunk") { + events.push(`chunk:${u.update.content?.text}`); + } + }, + } as unknown as AcpClient; + const agent = new ClaudeAcpAgent(mockClient, { log: () => {}, error: () => {} }); + return { agent, events }; + }; + + /** The issue-#453 cache-replay shape: zero output tokens, no streaming, + * the answer only on the result text. */ + const replayedResult = (text: string) => + resultMessage({ + result: text, + usage: { + input_tokens: 10, + output_tokens: 0, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + }, + }); + + const backgroundTasksChanged = (taskIds: string[]) => ({ + type: "system", + subtype: "background_tasks_changed", + tasks: taskIds.map((task_id) => ({ + task_id, + task_type: "local_agent", + description: "explore", + })), + uuid: randomUUID(), + session_id: "test-session", + }); + + /** A background Task/Agent-tool subagent starting (subagent_type set). */ + const subagentStarted = (taskId: string) => ({ + type: "system", + subtype: "task_started", + task_id: taskId, + tool_use_id: `toolu_${taskId}`, + description: "Explore the project", + subagent_type: "Explore", + uuid: randomUUID(), + session_id: "test-session", + }); + + const taskNotification = (taskId: string) => ({ + type: "system", + subtype: "task_notification", + task_id: taskId, + tool_use_id: `toolu_${taskId}`, + status: "completed", + output_file: "", + summary: "done", + uuid: randomUUID(), + session_id: "test-session", + }); + + const assistantText = (text: string) => ({ + type: "assistant", + parent_tool_use_id: null, + uuid: randomUUID(), + session_id: "test-session", + message: { + role: "assistant", + model: "claude-sonnet-4-5", + stop_reason: "end_turn", + usage: { + input_tokens: 1, + output_tokens: 1, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + }, + content: [{ type: "text", text }], + }, + }); + + it("holds the prompt open while a subagent is live and resolves at the followup's result", async () => { + const { agent, events } = chunkCapturingAgent(); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const { value: userMessage } = await iter.next(); + yield userEcho(userMessage); + yield running(); + yield subagentStarted("agent-1"); + // The user turn's terminal result: the subagent is still live, so the + // prompt must NOT resolve here. + yield resultMessage(); + // The CLI does not hold its trailing idle for background agents — + // it arrives immediately, while the subagent still runs. The hold + // must survive it. + yield idle(); + // The subagent finishes; the model wakes and streams the promised + // followup summary, which must land inside the still-open turn. + yield taskNotification("agent-1"); + yield assistantText("promised summary"); + yield resultMessage({ origin: { kind: "task-notification" } }); + yield idle(); + } + return messageGenerator(); + }); + + const response = await agent + .prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "explore" }] }) + .then((r) => { + events.push("resolved"); + return r; + }); + + expect(response.stopReason).toBe("end_turn"); + // The user turn's own usage — the task-notification followup's tokens are + // reported separately, not folded into the prompt response. + expect(response.usage?.totalTokens).toBe(15); + // The followup summary streamed BEFORE the prompt resolved, i.e. inside + // the turn, where every client is still listening. + const summaryIndex = events.indexOf("chunk:promised summary"); + expect(summaryIndex).toBeGreaterThanOrEqual(0); + expect(summaryIndex).toBeLessThan(events.indexOf("resolved")); + await agent.sessions["test-session"]?.consumer; + }); + + it("falls back to settling at an idle when no followup comes", async () => { + const agent = createMockAgent(); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const { value: userMessage } = await iter.next(); + yield userEcho(userMessage); + yield running(); + yield subagentStarted("agent-1"); + yield resultMessage(); + yield idle(); // the turn's own trailer — still waiting, must hold + yield taskNotification("agent-1"); // subagent done, but no followup + yield idle(); // nothing left to wait for — settles here + } + return messageGenerator(); + }); + + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "explore" }], + }); + expect(response.stopReason).toBe("end_turn"); + await agent.sessions["test-session"]?.consumer; + }); + + it("resolves at the result when the subagent already settled during the turn", async () => { + const agent = createMockAgent(); + let releaseIdle!: () => void; + const idleGate = new Promise((resolve) => (releaseIdle = resolve)); + let idleYielded = false; + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const { value: userMessage } = await iter.next(); + yield userEcho(userMessage); + yield running(); + yield subagentStarted("agent-1"); + yield taskNotification("agent-1"); // settled before the result + yield resultMessage(); + await idleGate; + idleYielded = true; + yield idle(); + } + return messageGenerator(); + }); + + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "quick" }], + }); + expect(response.stopReason).toBe("end_turn"); + expect(idleYielded).toBe(false); + releaseIdle(); + await agent.sessions["test-session"]?.consumer; + }); + + it("resolves at the result for non-subagent background tasks (run_in_background Bash)", async () => { + // A dev server can outlive every turn; deferring on it would only add + // settlement latency to each one. Only subagent tasks defer. + const agent = createMockAgent(); + let releaseIdle!: () => void; + const idleGate = new Promise((resolve) => (releaseIdle = resolve)); + let idleYielded = false; + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const { value: userMessage } = await iter.next(); + yield userEcho(userMessage); + yield running(); + yield { + type: "system", + subtype: "task_started", + task_id: "bash-1", + tool_use_id: "toolu_bash", + description: "npm run dev", + // no subagent_type: a background shell + uuid: randomUUID(), + session_id: "test-session", + }; + yield resultMessage(); + await idleGate; + idleYielded = true; + yield idle(); + } + return messageGenerator(); + }); + + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "start the dev server" }], + }); + expect(response.stopReason).toBe("end_turn"); + expect(idleYielded).toBe(false); + releaseIdle(); + await agent.sessions["test-session"]?.consumer; + }); + + it("settles a deferred turn 'cancelled' immediately at cancel()", async () => { + // During the hold the session is typically already idle (its trailer + // fired at the result), so the interrupt may never produce a fresh idle + // — cancel() must not wait for one (or for the force-cancel backstop). + const agent = createMockAgent(); + let releaseAfterCancel!: () => void; + const afterCancel = new Promise((resolve) => (releaseAfterCancel = resolve)); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const { value: userMessage } = await iter.next(); + yield userEcho(userMessage); + yield running(); + yield subagentStarted("agent-1"); + yield resultMessage(); // deferral point + yield idle(); // the turn's own trailer — the hold survives it + await afterCancel; // nothing more comes until after the cancel + yield idle(); + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "explore" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn?.deferredSettle); + await agent.cancel({ sessionId: "test-session" }); + + // Resolved by cancel() itself — no further stream message needed. The + // turn's own result already accumulated its usage; the cancelled settle + // must still report it (issue #844). + await expect(first).resolves.toEqual({ + stopReason: "cancelled", + usage: expect.objectContaining({ totalTokens: 15 }), + }); + releaseAfterCancel(); + await agent.sessions["test-session"]?.consumer; + }); + + it("does not fail a held turn when a followup errors while another subagent is live", async () => { + // A followup's is_error must never touch the user-turn lifecycle: the + // held turn's own result recorded a success. + const agent = createMockAgent(); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const { value: userMessage } = await iter.next(); + yield userEcho(userMessage); + yield running(); + yield subagentStarted("agent-1"); + yield subagentStarted("agent-2"); + yield resultMessage(); // defers on both + yield idle(); + yield taskNotification("agent-1"); + // agent-1's followup errors while agent-2 is still live. + yield resultMessage({ + origin: { kind: "task-notification" }, + is_error: true, + result: "followup blew up", + }); + yield idle(); + yield taskNotification("agent-2"); + yield resultMessage({ origin: { kind: "task-notification" } }); // settles + yield idle(); + } + return messageGenerator(); + }); + + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "explore" }], + }); + expect(response.stopReason).toBe("end_turn"); + await agent.sessions["test-session"]?.consumer; + }); + + it("does not read a followup's lagged trailing idle as the next prompt being abandoned", async () => { + // Settling at the followup's result unblocks the client right before the + // followup's own trailing idle; if that idle lags past the next prompt's + // echo it must be absorbed, not read as the fresh turn ending without a + // result (issue #825's false-fail path). + const agent = createMockAgent(); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield userEcho(u1.value); + yield running(); + yield subagentStarted("agent-1"); + yield resultMessage(); // turn 1 defers + yield idle(); // turn 1's trailer + yield taskNotification("agent-1"); + yield resultMessage({ origin: { kind: "task-notification" } }); // settles turn 1 + const u2 = await iter.next(); + yield userEcho(u2.value); // turn 2 activates... + yield idle(); // ...before the followup's lagged trailer arrives + yield resultMessage(); // turn 2's own result + yield idle(); + } + return messageGenerator(); + }); + + const first = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + expect(first.stopReason).toBe("end_turn"); + const second = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "second" }], + }); + expect(second.stopReason).toBe("end_turn"); + await agent.sessions["test-session"]?.consumer; + }); + + it("hands off a deferred turn with its recorded stop reason when the next prompt's echo arrives", async () => { + // The user moving on must not block behind a long-running subagent — the + // next echo settles the deferred turn — but the hand-off must report the + // outcome its result recorded, not rewrite it to end_turn. + const agent = createMockAgent(); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield userEcho(u1.value); + yield running(); + yield subagentStarted("agent-1"); + yield resultMessage({ stop_reason: "max_tokens" }); // turn 1 defers + const u2 = await iter.next(); + yield userEcho(u2.value); // turn 2's echo hands off turn 1 + // Turn 2 did not spawn agent-1, so it settles at its own result even + // though the subagent is still live — an earlier turn's long-running + // agent must not stall later prompts. + yield resultMessage(); + yield idle(); + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn?.deferredSettle); + const second = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "second" }], + }); + + await expect(first).resolves.toEqual(expect.objectContaining({ stopReason: "max_tokens" })); + await expect(second).resolves.toEqual(expect.objectContaining({ stopReason: "end_turn" })); + await agent.sessions["test-session"]?.consumer; + }); + + it("hands off a held turn when an echo-less command's result arrives (/context during a hold)", async () => { + // An echo-less queued command has no user echo to trigger the hand-off, + // so its result must do it: settle the held turn with ITS recorded + // outcome and promote the queued turn — not overwrite the held turn's + // outcome and leave the queued prompt hanging. + const agent = createMockAgent(); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield userEcho(u1.value); + yield running(); + yield subagentStarted("agent-1"); + yield resultMessage({ stop_reason: "max_tokens" }); // turn 1 held + yield idle(); + await iter.next(); // second prompt pushed — echo-less, only a result + yield resultMessage(); + yield idle(); + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "explore" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn?.deferredSettle); + const second = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "/context" }], + }); + + await expect(first).resolves.toEqual(expect.objectContaining({ stopReason: "max_tokens" })); + await expect(second).resolves.toEqual(expect.objectContaining({ stopReason: "end_turn" })); + await agent.sessions["test-session"]?.consumer; + }); + + it("defers a refusal result while the turn's subagent is live", async () => { + // A refusal is a normal turn outcome — it must route through the same + // deferral gate, or the subagent's remaining work is stranded + // out-of-turn through the refusal lane. + const agent = createMockAgent(); + let releaseDrain!: () => void; + const drainGate = new Promise((resolve) => (releaseDrain = resolve)); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const { value: userMessage } = await iter.next(); + yield userEcho(userMessage); + yield running(); + yield subagentStarted("agent-1"); + yield resultMessage({ stop_reason: "refusal" }); // held, not settled + yield idle(); + await drainGate; + yield taskNotification("agent-1"); + yield resultMessage({ origin: { kind: "task-notification" } }); // settles + yield idle(); + } + return messageGenerator(); + }); + + const response = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "explore" }], + }); + // The refusal result must HOLD the turn (deferredSettle recorded), not + // settle it out from under the live subagent. + await waitFor( + () => agent.sessions["test-session"]?.activeTurn?.deferredSettle?.stopReason === "refusal", + ); + releaseDrain(); + await expect(response).resolves.toEqual(expect.objectContaining({ stopReason: "refusal" })); + await agent.sessions["test-session"]?.consumer; + }); + + it("resolves a held turn with its recorded outcome when the stream dies", async () => { + // The held turn's answer already streamed; a stream death during the + // post-answer hold is a background failure, not the turn's. + const agent = createMockAgent(); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const { value: userMessage } = await iter.next(); + yield userEcho(userMessage); + yield running(); + yield subagentStarted("agent-1"); + yield resultMessage(); // held + yield idle(); + throw new Error("subprocess died"); + } + return messageGenerator(); + }); + + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "explore" }], + }); + expect(response.stopReason).toBe("end_turn"); + await agent.sessions["test-session"]?.consumer; + }); + + it("absorbs the interrupt's lagged trailer after cancelling a held turn mid-followup", async () => { + // Cancelling while a followup cycle is live pre-empts its result, so + // the interrupt produces a trailer idle with no counted result; lagging + // past the next prompt's echo it must be absorbed, not read as that + // fresh turn ending without a result (issue #825's false-fail). (An + // already-idle cancel produces no trailer, and pre-counting one there + // would mask a future #825 detection — hence the not-idle gate, which + // its own test pins below.) + const agent = createMockAgent(); + let releaseAfterCancel!: () => void; + const afterCancel = new Promise((resolve) => (releaseAfterCancel = resolve)); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield userEcho(u1.value); + yield running(); + yield subagentStarted("agent-1"); + yield resultMessage(); // turn 1 held + yield idle(); // its trailer, absorbed mid-hold + yield taskNotification("agent-1"); + yield running(); // the followup cycle starts... + await afterCancel; // ...and the cancel's interrupt pre-empts it + const u2 = await iter.next(); + yield userEcho(u2.value); // turn 2 activates... + yield idle(); // ...before the interrupt's lagged trailer arrives + yield resultMessage(); // turn 2's own result + yield idle(); + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "explore" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn?.deferredSettle); + await agent.cancel({ sessionId: "test-session" }); + await expect(first).resolves.toEqual(expect.objectContaining({ stopReason: "cancelled" })); + releaseAfterCancel(); + + const second = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "next" }], + }); + expect(second.stopReason).toBe("end_turn"); + await agent.sessions["test-session"]?.consumer; + }); + + it("reconciles a leaked subagent entry via background_tasks_changed", async () => { + // If a task's settle bookend is lost, the level signal's REPLACE + // semantics drop the stale entry so later turns don't defer forever. + const agent = createMockAgent(); + let releaseIdle!: () => void; + const idleGate = new Promise((resolve) => (releaseIdle = resolve)); + let idleYielded = false; + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const { value: userMessage } = await iter.next(); + yield userEcho(userMessage); + yield running(); + yield subagentStarted("agent-1"); + // The settle bookend was lost; the level signal says nothing is live. + yield backgroundTasksChanged([]); + yield resultMessage(); + await idleGate; + idleYielded = true; + yield idle(); + } + return messageGenerator(); + }); + + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "explore" }], + }); + expect(response.stopReason).toBe("end_turn"); + expect(idleYielded).toBe(false); + // The entry survives for permission attribution (a live sync subagent is + // legitimately absent from the level's background-only universe) — only + // the hold stops waiting on it. + const record = agent.sessions["test-session"]!.liveBackgroundTasks.get("agent-1"); + expect(record?.parentToolUseId).toBe("toolu_agent-1"); + expect(record?.endedPerLevel).toBe("ended"); + releaseIdle(); + await agent.sessions["test-session"]?.consumer; + }); + + it("keeps holding through a peer-origin autonomous result", async () => { + // A peer/coordinator cycle's result is not the user's; it must neither + // settle the held turn as a hand-off nor touch the user-turn lifecycle. + const { agent, events } = chunkCapturingAgent(); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const { value: userMessage } = await iter.next(); + yield userEcho(userMessage); + yield running(); + yield subagentStarted("agent-1"); + yield resultMessage(); // held + yield idle(); + // A peer message wakes the model; its cycle's result must not end + // the hold (the subagent is still live). + yield resultMessage({ origin: { kind: "peer", from: "other-session" } }); + yield idle(); + yield assistantText("peer-cycle marker"); + yield taskNotification("agent-1"); + yield resultMessage({ origin: { kind: "task-notification" } }); // settles + yield idle(); + } + return messageGenerator(); + }); + + const response = await agent + .prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "explore" }] }) + .then((r) => { + events.push("resolved"); + return r; + }); + + expect(response.stopReason).toBe("end_turn"); + // Resolution came after the peer cycle's output — the peer result did + // not settle the hold. + const markerIndex = events.indexOf("chunk:peer-cycle marker"); + expect(markerIndex).toBeGreaterThanOrEqual(0); + expect(markerIndex).toBeLessThan(events.indexOf("resolved")); + await agent.sessions["test-session"]?.consumer; + }); + + it("closes the delivery stretch when a held turn is handed off by the next echo", async () => { + // The held turn's followup summary latched the delivery flag; the echo + // hand-off settles the held turn, so the flag must reset or the next + // (replayed) turn's issue-#453 result-text fallback is suppressed. + // Asserted on the observable: the replayed turn's result text IS + // forwarded (a flag check after turn 2's finally would pass vacuously). + const { agent, events } = chunkCapturingAgent(); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield userEcho(u1.value); + yield running(); + yield subagentStarted("agent-1"); + yield subagentStarted("agent-2"); + yield resultMessage(); // held on both + yield idle(); + yield taskNotification("agent-1"); + yield assistantText("first summary"); // latches the delivery flag + yield resultMessage({ origin: { kind: "task-notification" } }); // still holding (agent-2) + yield idle(); + const u2 = await iter.next(); + yield userEcho(u2.value); // hand-off settles the held turn + // Turn 2 is a cache-replayed turn: nothing streams, zero output + // tokens, the answer only on the result text (issue #453's shape). + yield replayedResult("replayed answer"); + yield idle(); + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "explore" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn?.deferredSettle); + const second = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "next" }], + }); + + await expect(first).resolves.toEqual(expect.objectContaining({ stopReason: "end_turn" })); + await expect(second).resolves.toEqual(expect.objectContaining({ stopReason: "end_turn" })); + // The hand-off closed the held turn's stretch, so the replayed turn's + // fallback fired instead of being suppressed by the followup's text. + expect(events).toContain("chunk:replayed answer"); + await agent.sessions["test-session"]?.consumer; + }); + + it("re-arms the hold when a later level includes a previously absent task, and sweeps ended entries at activation", async () => { + const agent = createMockAgent(); + let releaseDrain!: () => void; + const drainGate = new Promise((resolve) => (releaseDrain = resolve)); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield userEcho(u1.value); + yield running(); + yield subagentStarted("agent-1"); + // A racing level omits the live agent (payload built before its + // registration) — marks it endedPerLevel... + yield backgroundTasksChanged([]); + // ...and a later level that INCLUDES it proves it live: un-ended, + // so the turn's result defers on it again. + yield backgroundTasksChanged(["agent-1"]); + yield resultMessage(); // must HOLD (the un-mark re-armed the wait) + yield idle(); + await drainGate; // let the test observe the held state + // A second racing level ends it again while held; nothing settles + // here (the level precedes the notification in the normal ordering). + yield backgroundTasksChanged([]); + yield idle(); // the drain fallback settles the now-unwaited hold + const u2 = await iter.next(); + yield userEcho(u2.value); // turn 2 activation ARMS the sweep + yield resultMessage(); + yield idle(); + const u3 = await iter.next(); + yield userEcho(u3.value); // turn 3 activation deletes the armed entry + yield resultMessage(); + yield idle(); + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "explore" }], + }); + // Deferral engaged despite the earlier absent-mark — the inclusion + // un-ended the entry. + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn?.deferredSettle); + releaseDrain(); + await expect(first).resolves.toEqual(expect.objectContaining({ stopReason: "end_turn" })); + + const second = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "next" }], + }); + expect(second.stopReason).toBe("end_turn"); + // The first activation only ARMS the sweep — the entry survives one full + // turn so a corrective inclusive level can still rescue a live agent's + // attribution (deletion is irreversible). + expect(agent.sessions["test-session"]!.liveBackgroundTasks.get("agent-1")?.endedPerLevel).toBe( + "sweep-armed", + ); + + const third = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "again" }], + }); + expect(third.stopReason).toBe("end_turn"); + // The second activation deletes it (growth bound for lost bookends). + expect(agent.sessions["test-session"]!.liveBackgroundTasks.has("agent-1")).toBe(false); + await agent.sessions["test-session"]?.consumer; + }); + + it("rescues a sweep-armed entry when an inclusive level arrives mid-grace", async () => { + // The grace's whole point: an entry armed at one activation must be + // rescued — attribution intact, sweep disarmed in the same assignment — + // by an inclusive level before the next activation deletes it. + const agent = createMockAgent(); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield userEcho(u1.value); + yield running(); + yield subagentStarted("agent-1"); + yield backgroundTasksChanged([]); // racing payload marks it ended + yield resultMessage(); // no hold (ended) — settles at the result + yield idle(); + const u2 = await iter.next(); + yield userEcho(u2.value); // activation arms the sweep + yield resultMessage(); + yield backgroundTasksChanged(["agent-1"]); // mid-grace rescue + yield idle(); + const u3 = await iter.next(); + yield userEcho(u3.value); // must NOT delete the rescued entry + yield resultMessage(); + yield idle(); + } + return messageGenerator(); + }); + + for (const text of ["one", "two", "three"]) { + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text }], + }); + expect(response.stopReason).toBe("end_turn"); + } + const record = agent.sessions["test-session"]!.liveBackgroundTasks.get("agent-1"); + expect(record?.parentToolUseId).toBe("toolu_agent-1"); + expect(record?.endedPerLevel).toBeUndefined(); + await agent.sessions["test-session"]?.consumer; + }); + + it("absorbs the trailer when cancelling a hold blocked on a permission request", async () => { + // requires_action is a live cycle too (a followup waiting on a + // permission decision — the very scenario users cancel out of); + // interrupting it produces a result-less trailer just like running. + const agent = createMockAgent(); + let releaseAfterCancel!: () => void; + const afterCancel = new Promise((resolve) => (releaseAfterCancel = resolve)); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield userEcho(u1.value); + yield running(); + yield subagentStarted("agent-1"); + yield resultMessage(); // turn 1 held + yield idle(); // trailer absorbed + yield taskNotification("agent-1"); + yield running(); + yield requiresAction(); // the followup blocks on a permission request + await afterCancel; // the cancel's interrupt pre-empts it + const u2 = await iter.next(); + yield userEcho(u2.value); // turn 2 activates... + yield idle(); // ...before the interrupt's lagged trailer arrives + yield resultMessage(); // turn 2's own result + yield idle(); + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "explore" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn?.deferredSettle); + await agent.cancel({ sessionId: "test-session" }); + await expect(first).resolves.toEqual(expect.objectContaining({ stopReason: "cancelled" })); + releaseAfterCancel(); + + const second = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "next" }], + }); + expect(second.stopReason).toBe("end_turn"); + await agent.sessions["test-session"]?.consumer; + }); + + it("closes the stretch after autonomous prose so a replayed prompt still delivers", async () => { + // Autonomous prose (a wake's summary) latches the delivery flag; with no + // turn in flight or queued, the autonomous result must close the stretch + // or the next cache-replayed prompt's issue-#453 fallback is suppressed. + const { agent, events } = chunkCapturingAgent(); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield userEcho(u1.value); + yield running(); + yield resultMessage(); // turn 1 settles normally + yield idle(); + // Autonomous cycle with no turn pending: prose + result. + yield assistantText("background note"); + yield resultMessage({ origin: { kind: "task-notification" } }); + yield idle(); + const u2 = await iter.next(); + yield userEcho(u2.value); + // Cache-replayed turn: no streaming, zero output tokens, the answer + // only on the result text. + yield replayedResult("replayed answer"); + yield idle(); + } + return messageGenerator(); + }); + + const first = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "one" }], + }); + expect(first.stopReason).toBe("end_turn"); + // The user types AFTER the background note arrived (the real sequence); + // prompting before the autonomous result is consumed would race the + // queued-turn guard, which deliberately errs toward suppression. Wait + // for the prose to land and the autonomous result's clear to follow. + await waitFor( + () => + events.includes("chunk:background note") && + agent.sessions["test-session"]!.emittedAssistantText === false, + ); + const second = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "two" }], + }); + expect(second.stopReason).toBe("end_turn"); + expect(events).toContain("chunk:replayed answer"); + await agent.sessions["test-session"]?.consumer; + }); + + it("preserves a queued turn's streamed answer across an interleaved autonomous result", async () => { + // Mid-message echo lag: a queued turn's deltas can stream before its echo + // activates it (activeTurn still null). An autonomous result landing in + // that window must NOT close the stretch — the flag guards the queued + // turn's already-delivered answer, and clearing would re-emit it via the + // issue-#453 fallback (the duplicate direction the flag's doc forbids). + const { agent, events } = chunkCapturingAgent(); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield userEcho(u1.value); + yield running(); + yield resultMessage(); // turn 1 settles + yield idle(); + const u2 = await iter.next(); // turn 2 pushed (queued, not yet echoed) + // Turn 2's answer streams BEFORE its echo (mid-message echo lag). + yield assistantText("streamed answer"); + // An autonomous result interleaves while activeTurn is null but + // turn 2 sits unsettled in the queue. + yield resultMessage({ origin: { kind: "task-notification" } }); + yield userEcho(u2.value); // now the echo activates turn 2 + // Zero-output result whose text duplicates the streamed answer. + yield replayedResult("streamed answer"); + yield idle(); + yield idle(); + } + return messageGenerator(); + }); + + const first = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "one" }], + }); + expect(first.stopReason).toBe("end_turn"); + const second = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "two" }], + }); + expect(second.stopReason).toBe("end_turn"); + // The streamed answer must appear exactly once — the queued-turn guard + // kept the flag latched, so the fallback did not re-emit it. + expect(events.filter((e) => e === "chunk:streamed answer")).toHaveLength(1); + await agent.sessions["test-session"]?.consumer; + }); + + it("does not pre-count an interrupt trailer when cancelling an already-idle hold", async () => { + // An already-idle hold's interrupt emits no trailer; a pre-counted debt + // would never drain and would absorb the one un-owed idle that is the + // issue-#825 detector's signal for a later genuinely-wedged turn. + const agent = createMockAgent(); + let releaseAfterCancel!: () => void; + const afterCancel = new Promise((resolve) => (releaseAfterCancel = resolve)); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield userEcho(u1.value); + yield running(); + yield subagentStarted("agent-1"); + yield resultMessage(); // turn 1 held + yield idle(); // trailer absorbed; session sits idle + await afterCancel; // cancel happens here — no cycle running + const u2 = await iter.next(); + yield userEcho(u2.value); // turn 2 activates + yield idle(); // un-owed idle with turn 2 unsettled — the #825 signal + yield resultMessage(); // never reached for turn 2's settle + yield idle(); + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "explore" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn?.deferredSettle); + await agent.cancel({ sessionId: "test-session" }); + await expect(first).resolves.toEqual(expect.objectContaining({ stopReason: "cancelled" })); + releaseAfterCancel(); + + // The #825 detection must still fire for turn 2 — a stale pre-counted + // debt from the idle cancel would have absorbed the un-owed idle. + await expect( + agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "next" }] }), + ).rejects.toMatchObject({ code: -32603 }); + await agent.sessions["test-session"]?.consumer; + }); +}); + describe("session/cancel wedge recovery (issue #680)", () => { function createMockAgent() { const mockClient = { @@ -8056,7 +9110,9 @@ describe("session/cancel wedge recovery (issue #680)", () => { taskState: new Map(), toolUseCache: {}, emittedToolCalls: new Set(), - subagentParentToolUseIds: new Map(), + liveBackgroundTasks: new Map(), + emittedAssistantText: false, + owedTrailingIdles: 0, messageIdToUuid: new Map(), }; return { interrupt }; @@ -9323,7 +10379,9 @@ describe("agent selection config option", () => { taskState: new Map(), toolUseCache: {}, emittedToolCalls: new Set(), - subagentParentToolUseIds: new Map(), + liveBackgroundTasks: new Map(), + emittedAssistantText: false, + owedTrailingIdles: 0, messageIdToUuid: new Map(), }; return { session: agent.sessions[sessionId]!, applyFlagSettings };