diff --git a/package-lock.json b/package-lock.json index 42f2befe..7ed5f76c 100644 --- a/package-lock.json +++ b/package-lock.json @@ -10,7 +10,7 @@ "license": "Apache-2.0", "dependencies": { "@agentclientprotocol/sdk": "1.2.1", - "@anthropic-ai/claude-agent-sdk": "0.3.205", + "@anthropic-ai/claude-agent-sdk": "0.3.206", "zod": "^3.25.0 || ^4.0.0" }, "bin": { @@ -26,7 +26,7 @@ "eslint": "10.6.0", "eslint-config-prettier": "10.1.8", "globals": "17.7.0", - "prettier": "3.9.4", + "prettier": "3.9.5", "ts-node": "10.9.2", "typescript": "6.0.3", "vitest": "4.1.10" @@ -45,22 +45,22 @@ } }, "node_modules/@anthropic-ai/claude-agent-sdk": { - "version": "0.3.205", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk/-/claude-agent-sdk-0.3.205.tgz", - "integrity": "sha512-ft6iBw9kXudsusiXNpeybIPBJ07Z3tqp1ROSg5cEJqgA+9i+JJj2sRfQth+QD+lyenbbAU8yPieLxIimvfBhtw==", + "version": "0.3.206", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk/-/claude-agent-sdk-0.3.206.tgz", + "integrity": "sha512-KljDh9Pg4YCYpoXS8dnWoVSsOHtU4yLCW268K2iOruSxFXE8/Tay6DPvmJzYuqjP5YLNYfj05ZGykwZSUn6GXA==", "license": "SEE LICENSE IN README.md", "engines": { "node": ">=18.0.0" }, "optionalDependencies": { - "@anthropic-ai/claude-agent-sdk-darwin-arm64": "0.3.205", - "@anthropic-ai/claude-agent-sdk-darwin-x64": "0.3.205", - "@anthropic-ai/claude-agent-sdk-linux-arm64": "0.3.205", - "@anthropic-ai/claude-agent-sdk-linux-arm64-musl": "0.3.205", - "@anthropic-ai/claude-agent-sdk-linux-x64": "0.3.205", - "@anthropic-ai/claude-agent-sdk-linux-x64-musl": "0.3.205", - "@anthropic-ai/claude-agent-sdk-win32-arm64": "0.3.205", - "@anthropic-ai/claude-agent-sdk-win32-x64": "0.3.205" + "@anthropic-ai/claude-agent-sdk-darwin-arm64": "0.3.206", + "@anthropic-ai/claude-agent-sdk-darwin-x64": "0.3.206", + "@anthropic-ai/claude-agent-sdk-linux-arm64": "0.3.206", + "@anthropic-ai/claude-agent-sdk-linux-arm64-musl": "0.3.206", + "@anthropic-ai/claude-agent-sdk-linux-x64": "0.3.206", + "@anthropic-ai/claude-agent-sdk-linux-x64-musl": "0.3.206", + "@anthropic-ai/claude-agent-sdk-win32-arm64": "0.3.206", + "@anthropic-ai/claude-agent-sdk-win32-x64": "0.3.206" }, "peerDependencies": { "@anthropic-ai/sdk": ">=0.93.0", @@ -69,9 +69,9 @@ } }, "node_modules/@anthropic-ai/claude-agent-sdk-darwin-arm64": { - "version": "0.3.205", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-darwin-arm64/-/claude-agent-sdk-darwin-arm64-0.3.205.tgz", - "integrity": "sha512-lrfJ4eVtzfPkCpbSkBOGSMQCBbvmW6nbPzgHE4IwMN3scZlpuFMUFqh2aaJa/X2SAcWD9H2S0t2WWvSRgM7BjA==", + "version": "0.3.206", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-darwin-arm64/-/claude-agent-sdk-darwin-arm64-0.3.206.tgz", + "integrity": "sha512-FL2+NKcMMN47vcnCW2Fkt3AOgeRRlQxrisbPNaxrxqPJFzhUKs17x5j0XzLefd0xRbDAr74hd0PK/tnp6PHM6w==", "cpu": [ "arm64" ], @@ -82,9 +82,9 @@ ] }, "node_modules/@anthropic-ai/claude-agent-sdk-darwin-x64": { - "version": "0.3.205", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-darwin-x64/-/claude-agent-sdk-darwin-x64-0.3.205.tgz", - "integrity": "sha512-G6ETPmL5mNzJ2DFsWxG3jmsmrXgZX1N2ZCJvxaGUUpjTsKZJ4Tup1cWYvcd/m7o5fYZmx9REmgzTwsAIc1fdPQ==", + "version": "0.3.206", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-darwin-x64/-/claude-agent-sdk-darwin-x64-0.3.206.tgz", + "integrity": "sha512-mRW8PPMfQN15EunLwpdmcVzk3XuM4DXQUM8DOzaeA1Hr1yxYPaVVBLr9hkdBKOqr8XTl3ueoTR6RZAy6a/n7OA==", "cpu": [ "x64" ], @@ -95,9 +95,9 @@ ] }, "node_modules/@anthropic-ai/claude-agent-sdk-linux-arm64": { - "version": "0.3.205", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-linux-arm64/-/claude-agent-sdk-linux-arm64-0.3.205.tgz", - "integrity": "sha512-CXzySK3PV3EizCRPXnxPqeaAtgrBFDnMFOVpMe36oC3U16yDb1b1tAJGqZi/7uFrVvAiaXvnSFxhUWnDDSaO+A==", + "version": "0.3.206", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-linux-arm64/-/claude-agent-sdk-linux-arm64-0.3.206.tgz", + "integrity": "sha512-Id6H8l6EsGb7849EAZDOB4Ic+FQpbzt4D5HGOwp59CTW3o/1c4etjQA/Kl1K+DSEWn3FGo6D5UMVgc/rwhBj8g==", "cpu": [ "arm64" ], @@ -111,9 +111,9 @@ ] }, "node_modules/@anthropic-ai/claude-agent-sdk-linux-arm64-musl": { - "version": "0.3.205", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-linux-arm64-musl/-/claude-agent-sdk-linux-arm64-musl-0.3.205.tgz", - "integrity": "sha512-91fgdG4aTnQ29sKOcUqgH4+tKCW2ut6PWGRSYmXNDbROasJm1rAlPdzC5brdu/e4c0CDSNV6TWyE5JCjaS/jlQ==", + "version": "0.3.206", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-linux-arm64-musl/-/claude-agent-sdk-linux-arm64-musl-0.3.206.tgz", + "integrity": "sha512-aMZe1Kl+kYv5QlA15W9Ae25MAzBsA9FA40f5TOtJebA+M/xliF0r2LLb2NdyBviiZCDllcE31l0zgYE0rQqFQw==", "cpu": [ "arm64" ], @@ -127,9 +127,9 @@ ] }, "node_modules/@anthropic-ai/claude-agent-sdk-linux-x64": { - "version": "0.3.205", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-linux-x64/-/claude-agent-sdk-linux-x64-0.3.205.tgz", - "integrity": "sha512-siS+1iNqBSlGFZZvJY6+mhzZ/6/ec/TbX9GMuwmTF0E6fxGhIIp797jJxR1q8r6FAq7d39mEoRNhC0Ffo60uNQ==", + "version": "0.3.206", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-linux-x64/-/claude-agent-sdk-linux-x64-0.3.206.tgz", + "integrity": "sha512-egZhOC1RlEVhZyq6Oa1b04AF7hh4hO+8oCsJCZ3gOifJQuHih1oW3lZ8fOUxU85jlT2ytAnt5kRp4uQr6PJgbg==", "cpu": [ "x64" ], @@ -143,9 +143,9 @@ ] }, "node_modules/@anthropic-ai/claude-agent-sdk-linux-x64-musl": { - "version": "0.3.205", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-linux-x64-musl/-/claude-agent-sdk-linux-x64-musl-0.3.205.tgz", - "integrity": "sha512-vvsb7GlnA8CTSVvvTkrXjcSeRKqxSM7p/tU3Od9ICAZeWHglptekEyzLEApzLuLbI5ewfFF/F0q3NwOBbo18dg==", + "version": "0.3.206", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-linux-x64-musl/-/claude-agent-sdk-linux-x64-musl-0.3.206.tgz", + "integrity": "sha512-xQBOBhlcmTNc7YeYT5qLZOikpSq3WHlsJ/t6i7kJqUWrcXlOPaAoSMdKp5xO/V0CjAdzdJoWB88I55UAh8mclQ==", "cpu": [ "x64" ], @@ -159,9 +159,9 @@ ] }, "node_modules/@anthropic-ai/claude-agent-sdk-win32-arm64": { - "version": "0.3.205", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-win32-arm64/-/claude-agent-sdk-win32-arm64-0.3.205.tgz", - "integrity": "sha512-SpP5zF68weFez/6pKrGzq/UVAJDMDNphWqmkLfOpWTDBL5xy6XlIZw5Bl4EXoVnfi2VLFkwuffNeFe+9SdX7kw==", + "version": "0.3.206", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-win32-arm64/-/claude-agent-sdk-win32-arm64-0.3.206.tgz", + "integrity": "sha512-oRQk23bFXSz4QRhOxqnnvlLWq/KiF2PtSBYXrMg1AQrCOHJd2k66aOAS7AJ4ZSzejiyV4N6sS6UThfmF6ipHBQ==", "cpu": [ "arm64" ], @@ -172,9 +172,9 @@ ] }, "node_modules/@anthropic-ai/claude-agent-sdk-win32-x64": { - "version": "0.3.205", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-win32-x64/-/claude-agent-sdk-win32-x64-0.3.205.tgz", - "integrity": "sha512-kg2kkXyeSoFLruO3Ic2IruLxzBR0xCUtmlJHdWi3SYW7JhAKNJg4fcrdJsWcardmEw23Y2UDGDJbRyxqSVx6wg==", + "version": "0.3.206", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-win32-x64/-/claude-agent-sdk-win32-x64-0.3.206.tgz", + "integrity": "sha512-BdjKmDojZjc5RjN+8Q6K7Yqf1WYhel6I3DkMXc9zdxS/xIuN42sx7zgQpdMGY73pm7ROPLqo3LieOeMhVH161w==", "cpu": [ "x64" ], @@ -2371,9 +2371,9 @@ } }, "node_modules/hono": { - "version": "4.12.28", - "resolved": "https://registry.npmjs.org/hono/-/hono-4.12.28.tgz", - "integrity": "sha512-YwUvVpSF7m1yOblFPrU3Hbo8XhPheBoiyfGuII6z19LnOr6JpDnyyp7LFNrfV56wS8tpvtBFGRISHN02pDdLOA==", + "version": "4.12.29", + "resolved": "https://registry.npmjs.org/hono/-/hono-4.12.29.tgz", + "integrity": "sha512-1hNiRjawYrLq/4m3DQQjPGFg0VZkk4RjQJDff/excI6Dm9BiL75qxGrd7/c6YOxPdq6AscP3LiXhQ6fKFC1Waw==", "license": "MIT", "peer": true, "engines": { @@ -3227,9 +3227,9 @@ } }, "node_modules/prettier": { - "version": "3.9.4", - "resolved": "https://registry.npmjs.org/prettier/-/prettier-3.9.4.tgz", - "integrity": "sha512-yWG/o/4oJfo036EKAfK6ACAoDOfHeRHx4tuxkfBZiauURiaSmYwlpOr5LQqKtIkRD2z1PLteme2WoxEnj4tHTg==", + "version": "3.9.5", + "resolved": "https://registry.npmjs.org/prettier/-/prettier-3.9.5.tgz", + "integrity": "sha512-/FVl766LpUfB5vXgCYOYa0MeV/441Ia99AeICQIQFTY/Nw0roZwULcXpku5i1/m5kt/baz+s4Zogspd839HSMg==", "dev": true, "license": "MIT", "bin": { diff --git a/package.json b/package.json index b4c4dd92..dd563650 100644 --- a/package.json +++ b/package.json @@ -64,7 +64,7 @@ }, "dependencies": { "@agentclientprotocol/sdk": "1.2.1", - "@anthropic-ai/claude-agent-sdk": "0.3.205", + "@anthropic-ai/claude-agent-sdk": "0.3.206", "zod": "^3.25.0 || ^4.0.0" }, "devDependencies": { @@ -77,7 +77,7 @@ "eslint": "10.6.0", "eslint-config-prettier": "10.1.8", "globals": "17.7.0", - "prettier": "3.9.4", + "prettier": "3.9.5", "ts-node": "10.9.2", "typescript": "6.0.3", "vitest": "4.1.10" diff --git a/src/acp-agent.ts b/src/acp-agent.ts index 25d7c552..6b48f7ac 100644 --- a/src/acp-agent.ts +++ b/src/acp-agent.ts @@ -209,6 +209,30 @@ type Turn = { /** Set once the deferred has been resolved/rejected, so the consumer never * settles a turn twice (idle + handoff + stream-end can all race). */ settled: boolean; + /** Set when a `command_lifecycle` "started" frame arrives for this turn's + * uuid (msg_lifecycle_v1 CLIs): the SDK dispatched the command into a turn. + * Read by cancel() to seed the orphan's state — a started orphan's turn may + * still emit a result, an undispatched one may be dropped without one. */ + commandStarted?: boolean; + /** Set when a terminal `command_lifecycle` frame arrives for this turn's + * uuid while the turn is still queued (msg_lifecycle_v1 CLIs). The command + * is already finished SDK-side, so a later cancel() must not seed an + * orphan entry for it — no terminal frame will ever come to drain it. + * "completed"/"discarded" leave nothing outstanding; "cancelled" after a + * dispatch means the dead turn's result may still arrive (seeded as a + * zombie) unless it already passed (`commandResultSeen`), and without a + * dispatch means dropped (nothing coming). */ + commandFinished?: "completed" | "discarded" | "cancelled"; + /** Set when a user-turn result arrives while this command is known + * dispatched (`commandStarted`) with no terminal frame yet. Turns run + * sequentially and frames arrive in stream order, so the turn this command + * was dispatched into IS the turn that emitted that result — including + * when the command was FOLDED into another turn (their shared result). + * Read by cancel() and the force-cancel wedge path so neither seeds an + * orphan entry for a result that has already passed: such an entry could + * never be drained by its result and would swallow an unrelated later + * echo-less one instead. */ + commandResultSeen?: boolean; resolve: (response: PromptResponse) => void; reject: (error: unknown) => void; }; @@ -233,8 +257,40 @@ type Session = { * the interrupt dropped (absent from `still_queued`) are uncounted as soon * as the receipt arrives (see cancel()). Reset to 0 on every activation as * a backstop against a dropped queued input this can't see (older CLIs, a - * receipt lost to a failed control round-trip). */ + * receipt lost to a failed control round-trip). Only used when the CLI does + * NOT emit lifecycle frames (see `orphanCommands` for the msg_lifecycle_v1 + * lane); a count can't express command coalescing — N queued commands can + * fold into ONE turn emitting one result, leaving a stale skip of N-1. */ pendingOrphanResults?: number; + /** msg_lifecycle_v1 lane of the orphan accounting (see + * `pendingOrphanResults` for the count lane): the uuids of cancelled queued + * turns whose SDK-side command may still produce an unaccounted result, + * keyed to what we know of its fate. "pending" = not seen dispatched; if + * the SDK drops it (interrupt, `cancelled` before "started") no result + * ever comes. "started" = dispatched into a turn whose result is still + * coming; exactly one terminal lifecycle frame will follow. "zombie" = its + * turn was aborted/failed after dispatch with no result seen since + * (`cancelled` after "started"); no more lifecycle frames come, but the + * dead turn's error result may still arrive. Entries are removed the + * moment their result is covered: EVERY user-turn result covers ALL + * started and zombie entries at once (turns run sequentially and frames + * arrive in stream order, so at any result the started entries were + * dispatched into — possibly folded into — the emitting turn, and any + * zombie's late result has already passed or never existed), whether that + * result was attributed to the active turn or skipped echo-less (see + * recordResultForOrphanCommands / ensureActiveTurn). A command's own + * terminal frame also drains its entry ("completed" is emitted after any + * result its turn produced; a bare `cancelled` deletes a pending entry — + * dropped without running — and zombifies a started one). An echo-less + * result is an orphan's iff this map is non-empty (FIFO: orphan turns run + * before any live turn's). Cleared on every activation, same self-heal as + * the count (covers a lost frame, which can leak an entry — each state + * bounds the damage to one wrong skip). */ + orphanCommands?: Map; + /** True once a `system`/init advertised the msg_lifecycle_v1 capability, so + * cancel() routes orphan accounting to `orphanCommands` (exact, per-uuid) + * instead of `pendingOrphanResults` (count, coalescing-blind). */ + msgLifecycleV1?: boolean; /** The long-lived consumer task. Lazily started on the first `prompt()` and * kept alive for the session so between-turn/background messages are still * drained and forwarded. */ @@ -1235,6 +1291,7 @@ export class ClaudeAcpAgent { session.activeTurn = turn; session.cancelled = false; session.pendingOrphanResults = 0; + session.orphanCommands?.clear(); resetTurnScratch(); }; @@ -1257,17 +1314,95 @@ export class ClaudeAcpAgent { if (session.activeTurn) { return; } - const head = (session.turnQueue ?? []).find((t) => !t.settled); - if (!head) { - return; - } + // Orphan accounting runs BEFORE the head check: an orphan's echo-less + // result can arrive with an EMPTY queue (the common post-cancel + // timeline — the active turn settled at the interrupt's idle and the + // user hasn't typed yet), and it must still be consumed here. Skipping + // the bookkeeping when there is nothing to promote would leave a + // phantom entry/count that swallows the next live echo-less result + // (e.g. /compact) instead. if ((session.pendingOrphanResults ?? 0) > 0) { session.pendingOrphanResults!--; return; } + // msg_lifecycle_v1 lane. Attribute this echo-less result using the + // entries' states — turns run sequentially and frames arrive in stream + // order, so at any result: every "zombie" is from an already-dead turn + // whose own result already passed before the frame that created the + // newest entry (or never existed), every "started" entry was dispatched + // into THE turn that emitted this result (an older turn's entries got + // their terminal frames before a newer turn's "started" frames), and a + // "pending" entry was not dispatched before it. One result therefore + // covers ALL started and zombie entries at once (N coalesced commands + // share ONE result); their outstanding terminal frames then no-op on + // the missing entries. NOTE this ordering argument is asserted from + // observed CLI behavior, not a documented wire contract — if a dead + // turn's late result could lag past the NEXT turn's dispatch frames, + // deleting a zombie and a started entry on one result would + // double-consume it. The unexpected-transition logging in the frame + // handler is the tripwire for that class of drift. + if (session.orphanCommands?.size) { + let consumedOrphanResult = false; + let oldestPending: string | undefined; + for (const [uuid, state] of session.orphanCommands) { + if (state === "started" || state === "zombie") { + consumedOrphanResult = true; + session.orphanCommands.delete(uuid); + } else { + oldestPending ??= uuid; + } + } + if (consumedOrphanResult) { + return; + } + if (oldestPending !== undefined) { + // No dispatch was seen before this result, so it is very likely a + // live turn's — but a lost "started" frame would mean it IS the + // orphan's (dup-over-loss: prefer one wrong skip over + // misattributing a dead turn's outcome to a live prompt). Grant + // each pending entry exactly one skip, like the count lane did. + session.orphanCommands.delete(oldestPending); + return; + } + } + const head = (session.turnQueue ?? []).find((t) => !t.settled); + if (!head) { + return; + } activateTurn(head); }; + /** Result-time bookkeeping that must run whether or not the result can be + * attributed to a turn. (1) Latch `commandResultSeen` on every queued + * turn whose command is known dispatched with no terminal frame yet — + * the emitting turn is the one it was dispatched (possibly folded) into, + * so its result has now passed; a later cancel() must not seed an orphan + * entry that waits for it (see Turn.commandResultSeen). (2) When a turn + * is ACTIVE, the result is attributed to it and never reaches + * ensureActiveTurn — but it still covers the map's started entries + * (commands folded into the active turn share its result) and zombies + * (their late results have already passed or never existed), so drain + * them here or they would zombify/linger and swallow a later live + * echo-less result. */ + const recordResultForOrphanCommands = () => { + for (const turn of session.turnQueue ?? []) { + if (!turn.settled && turn.commandStarted && !turn.commandFinished) { + turn.commandResultSeen = true; + } + } + if (session.activeTurn && session.orphanCommands?.size) { + for (const [uuid, state] of session.orphanCommands) { + if (state === "started" || state === "zombie") { + session.orphanCommands.delete(uuid); + } + } + } + }; + + /** The unsettled in-flight turn owning this prompt uuid, if any. */ + const findUnsettledTurn = (uuid: string) => + (session.turnQueue ?? []).find((t) => t.promptUuid === uuid && !t.settled); + /** 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) => { @@ -1362,7 +1497,34 @@ export class ClaudeAcpAgent { // turn being abandoned. Stale counts self-heal: activation resets // them (see activateTurn). if (session.activeTurn && !session.activeTurn.settled) { - session.pendingOrphanResults = (session.pendingOrphanResults ?? 0) + 1; + // Seed by what the frames already told us, mirroring cancel()'s + // queued-turn sweep — the consumer may have drained the wedged + // turn's result and/or terminal frame before the backstop fired, + // and an entry seeded for a result or frame that is already + // spent would never drain (it would swallow an unrelated later + // echo-less result instead). + const active = session.activeTurn; + if (active.commandFinished === "completed" || active.commandFinished === "discarded") { + // Finished SDK-side; any result already passed. Nothing to + // track. + } else if (active.commandFinished === "cancelled") { + // Aborted after dispatch: its late result may still come — + // unless it already did. + if (!active.commandResultSeen) { + this.trackOrphanCommand(session, active.promptUuid, "zombie"); + } + } else if (active.commandResultSeen) { + // Its result was already consumed (dropped at the cancelled + // guard); only the terminal frame is outstanding, which no-ops + // with no entry. Nothing to track. + } else { + // The wedged turn WAS dispatched (it's active), so track it + // "started": its late result (if the SDK recovers) is skipped + // echo-less, and its terminal frame — or that skip plus + // activation's clear when the frame is lost to the wedge — is + // what drains it. + this.trackOrphanCommand(session, active.promptUuid, "started"); + } } settleActive({ stopReason: "cancelled", usage: sessionUsage(session) }); // If the session is being torn down, abandon the in-flight next() @@ -1426,10 +1588,106 @@ export class ClaudeAcpAgent { }); } + // CLIs 2.1.206+ (capability msg_lifecycle_v1) report the fate of every + // uuid-stamped queued command (queued/started/completed/cancelled/ + // discarded) as `command_lifecycle` frames — 2-3 per prompt, since + // prompt() stamps a uuid on every message. The frame is @internal and + // absent from the SDKMessage union, so handle it BEFORE the exhaustive + // switch: it must not reach `unreachable`'s error log, and a `case` + // for it wouldn't typecheck. It feeds only the orphan accounting (see + // Session.orphanCommands); turn settlement stays driven by + // echoes/results/idle. (Raw-mode emission above still forwards these + // frames.) + if ((message as { type: string }).type === "command_lifecycle") { + const frame = message as unknown as { command_uuid: string; state: string }; + switch (frame.state) { + case "started": { + // Remember dispatch on the live turn so a cancel() that orphans + // it seeds the right state (see Turn.commandStarted)... + const queued = findUnsettledTurn(frame.command_uuid); + if (queued) { + queued.commandStarted = true; + } + // ...and promote an already-orphaned command: once dispatched, + // a bare `cancelled` no longer means "dropped without running". + const state = session.orphanCommands?.get(frame.command_uuid); + if (state === "pending") { + session.orphanCommands!.set(frame.command_uuid, "started"); + } else if (state === "zombie") { + // "started" after the command's terminal frame: the ordering + // the whole lane rests on has been violated (frames are + // per-uuid FIFO). Surface it — a silent drift here degrades + // into swallowed or misattributed results. + this.logger.error( + `Session ${params.sessionId}: command_lifecycle "started" for ${frame.command_uuid} after its terminal frame; orphan accounting may be off for this cancel.`, + ); + } + break; + } + case "completed": + case "discarded": + case "cancelled": { + // Terminal frames. Latch the fate on a still-queued turn so a + // later cancel() doesn't seed an orphan entry for a command + // whose one-and-only terminal frame has already been consumed + // (nothing would ever drain that entry). + const queued = findUnsettledTurn(frame.command_uuid); + if (queued) { + queued.commandFinished = frame.state as "completed" | "discarded" | "cancelled"; + } + if (frame.state === "cancelled") { + // Ambiguous by design (dup-over-loss): dropped before + // dispatch (no result will ever come — safe to forget) vs + // consumed into a turn that was aborted/failed. For the + // latter, any result the dead turn managed to emit has + // already deleted the entry (see + // recordResultForOrphanCommands / ensureActiveTurn), so a + // still-"started" entry means no result was seen since + // dispatch — it becomes a zombie for the next + // echo-less-result skip. + const state = session.orphanCommands?.get(frame.command_uuid); + if (state === "pending") { + session.orphanCommands?.delete(frame.command_uuid); + } else if (state === "started") { + session.orphanCommands?.set(frame.command_uuid, "zombie"); + } + break; + } + // Exactly-one-terminal: the command is finished. "completed" is + // emitted after any result its turn produced (fresh turn) or the + // command folded into another turn whose result is attributed + // elsewhere — either way no echo-less result remains to skip. + // "discarded" = session ended with it still queued; no result. + session.orphanCommands?.delete(frame.command_uuid); + break; + } + default: + // "queued" carries no fate information. Anything else is a + // state this adapter doesn't know — likely a CLI that grew the + // v1 vocabulary. The entry still drains by result coverage or + // activation's clear (bounded damage), but log it so the + // degradation is visible instead of silent. + if (frame.state !== "queued") { + this.logger.error( + `Session ${params.sessionId}: unknown command_lifecycle state "${frame.state}" for ${frame.command_uuid}; treating as uninformative.`, + ); + } + break; + } + continue; + } + switch (message.type) { case "system": switch (message.subtype) { case "init": + // Latch the lifecycle capability so cancel() routes orphan + // accounting through `orphanCommands` (per-uuid, exact) + // instead of the coalescing-blind count. Never unlatch: init + // re-emits per turn and the capability can't be lost mid-CLI. + if (message.capabilities?.includes("msg_lifecycle_v1")) { + session.msgLifecycleV1 = true; + } // A fresh `system`/init (e.g. after reinitialize) can carry an // updated Fast mode state; reconcile it with what we seeded at // session creation. @@ -1863,7 +2121,12 @@ export class ClaudeAcpAgent { // no user-message echo to promote them, so do it here from the head. // Promote BEFORE accumulating usage, since activation resets the // accumulator — promoting after would discard this result's tokens. + // The orphan bookkeeping runs first: it covers folded/zombie + // 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) { + recordResultForOrphanCommands(); ensureActiveTurn(); } @@ -2187,9 +2450,7 @@ export class ClaudeAcpAgent { // is still promoted — activateTurn() clears the flag. The turn's own // echo is then dropped from the feed (the client already shows it). if (message.type === "user" && "uuid" in message && message.uuid) { - const queued = (session.turnQueue ?? []).find( - (t) => t.promptUuid === message.uuid && !t.settled, - ); + const queued = findUnsettledTurn(message.uuid); if (queued) { // Only (re)activate if this isn't already the active turn — a // turn promoted early (e.g. by a result that preceded its echo) @@ -2455,7 +2716,7 @@ export class ClaudeAcpAgent { case "conversation_reset": break; default: - unreachable(message); + unreachable(message, this.logger); break; } } @@ -2497,6 +2758,34 @@ export class ClaudeAcpAgent { } } + /** Route one orphaned command into the session's orphan-accounting lane: + * the per-uuid map on msg_lifecycle_v1 CLIs (drained by the command's own + * terminal lifecycle frame and the echo-less-result skip), the plain count + * elsewhere (the count lane can't express per-command states, so `state` + * only matters on the map lane). Both orphan-producing paths — cancel()'s + * queued-turn sweep and the consumer's force-cancel wedge path — must seed + * through here so the lane split stays a single mechanism. + * + * Known window: `msgLifecycleV1` is only learnable from the stream's first + * `system`/init (the control-channel initialize carries no capabilities), + * so a cancel that beats that drain seeds the COUNT lane on a + * lifecycle-capable CLI — where command coalescing can leave the count + * stale by N-1 (the pre-map bug, confined to this sub-second window and + * still healed by the next activation's reset). Structural until the SDK + * exposes capabilities before the stream starts. */ + private trackOrphanCommand( + session: Session, + uuid: string, + state: "pending" | "started" | "zombie", + ): void { + if (session.msgLifecycleV1) { + session.orphanCommands ??= new Map(); + session.orphanCommands.set(uuid, state); + } else { + session.pendingOrphanResults = (session.pendingOrphanResults ?? 0) + 1; + } + } + async cancel(params: CancelNotification): Promise { const session = this.sessions[params.sessionId]; if (!session) { @@ -2510,11 +2799,18 @@ export class ClaudeAcpAgent { return; } session.cancelled = true; + // Capture the orphan-accounting lane before anything can await: the + // consumer latches msgLifecycleV1 when it drains the first system/init, + // which can happen DURING the awaited interrupt() below — the receipt + // reconciliation must act on the same lane the seeding used, or a + // count-lane orphan would be left for the map-lane receipt path (which + // never decrements the count) to miss. + const lifecycleLane = session.msgLifecycleV1 === true; // Settle queued turns that haven't started yet (no echo seen) right away — // they have no in-flight SDK work to interrupt. The active turn is settled // by the consumer when it observes the interrupt's trailing idle (or via the // backstop below). Mirrors the old pendingMessages cancellation. - const orphanedUuids: string[] = []; + const orphanedTurns: Turn[] = []; if (session.turnQueue) { for (const turn of session.turnQueue) { if (turn !== session.activeTurn && !turn.settled) { @@ -2522,14 +2818,52 @@ export class ClaudeAcpAgent { // Deliberately no `usage`: a queued turn never ran, so the session // accumulator (the active turn's tally) is not its spend. turn.resolve({ stopReason: "cancelled" }); - orphanedUuids.push(turn.promptUuid); + orphanedTurns.push(turn); } } // Each removed queued turn's user message was already pushed to the SDK, // which processes input FIFO and will still emit a result for it with no - // uuid to match. Count those so the consumer skips them (see + // uuid to match. Track those so the consumer skips them (see // ensureActiveTurn) rather than misattributing them to the head. - session.pendingOrphanResults = (session.pendingOrphanResults ?? 0) + orphanedUuids.length; + // msg_lifecycle_v1 CLIs get per-uuid tracking drained by the command's + // own terminal lifecycle frame — exact under command coalescing, where + // N queued commands fold into ONE turn emitting one result and a plain + // count would go stale by N-1 and swallow a later echo-less result. + // Older CLIs keep the count and its activation-time self-heal (they + // never see lifecycle frames, so commandStarted/commandFinished stay + // unset and every turn takes the plain-seed path below). + for (const turn of orphanedTurns) { + if (turn.commandFinished === "completed" || turn.commandFinished === "discarded") { + // The command already finished SDK-side and its terminal frame was + // consumed while the turn sat queued — nothing is left to skip, and + // a seeded entry would never drain. + continue; + } + if (turn.commandFinished === "cancelled") { + // Terminal frame already consumed. Dispatched-then-aborted: the + // dead turn's late result may still come — seed the zombie the + // frame handler would have made — unless that result already + // passed pre-cancel (commandResultSeen: e.g. the command folded + // into the active turn and their shared result was attributed + // there), in which case a zombie would be a phantom that swallows + // an unrelated later result. Never dispatched: dropped, no result + // coming, nothing to track. + if (turn.commandStarted && !turn.commandResultSeen) { + this.trackOrphanCommand(session, turn.promptUuid, "zombie"); + } + continue; + } + if (turn.commandStarted && turn.commandResultSeen) { + // Dispatched and its turn's result already passed; only its + // terminal frame is outstanding, which no-ops with no entry. + continue; + } + this.trackOrphanCommand( + session, + turn.promptUuid, + turn.commandStarted ? "started" : "pending", + ); + } session.turnQueue = session.turnQueue.filter( (turn) => turn === session.activeTurn && !turn.settled, ); @@ -2577,11 +2911,31 @@ export class ClaudeAcpAgent { // receipt, so a bare `{}` success from a gateway can't read as "everything // was dropped") — keep the count-everything behavior and its // activation-time self-heal. - if (Array.isArray(receipt?.still_queued) && orphanedUuids.length > 0) { + if (Array.isArray(receipt?.still_queued) && orphanedTurns.length > 0) { const stillQueued = new Set(receipt.still_queued); - const dropped = orphanedUuids.filter((uuid) => !stillQueued.has(uuid)).length; - if (dropped > 0) { - session.pendingOrphanResults = Math.max(0, (session.pendingOrphanResults ?? 0) - dropped); + if (lifecycleLane) { + // Lifecycle lane: forget dropped orphans by uuid. Only entries still + // "pending" — an orphan absent from `still_queued` because it was + // DISPATCHED before the interrupt (not dropped) has usually been + // promoted to "started" by its lifecycle frame by now, and its own + // terminal frame must stay in charge of its fate. (If that frame is + // still in the consumer's backlog we mis-forget — the same exposure + // the count lane has always had for a dropped-then-run command.) + // Mostly redundant with the "cancelled"-frame removal, but a receipt + // survives paths where that frame was never emitted. + for (const turn of orphanedTurns) { + if ( + !stillQueued.has(turn.promptUuid) && + session.orphanCommands?.get(turn.promptUuid) === "pending" + ) { + session.orphanCommands.delete(turn.promptUuid); + } + } + } else { + const dropped = orphanedTurns.filter((turn) => !stillQueued.has(turn.promptUuid)).length; + if (dropped > 0) { + session.pendingOrphanResults = Math.max(0, (session.pendingOrphanResults ?? 0) - dropped); + } } } } diff --git a/src/tests/acp-agent.test.ts b/src/tests/acp-agent.test.ts index 7b348ef5..c39c9e16 100644 --- a/src/tests/acp-agent.test.ts +++ b/src/tests/acp-agent.test.ts @@ -82,6 +82,29 @@ function userEcho(u: any) { }; } +/** A `system`/init frame advertising the msg_lifecycle_v1 capability, so the + * consumer latches `session.msgLifecycleV1` and cancel() routes orphan + * accounting through `orphanCommands` (CLIs 2.1.206+). */ +const lifecycleInit = { + type: "system", + subtype: "init", + session_id: "test-session", + capabilities: ["interrupt_receipt_v1", "msg_lifecycle_v1"], +}; + +/** Build a `command_lifecycle` frame (CLIs 2.1.206+) reporting `state` for the + * uuid-stamped command `commandUuid`. One builder for every test so the + * @internal wire shape can't drift between suites. */ +function lifecycleFrame(commandUuid: string, state: string) { + return { + type: "command_lifecycle", + command_uuid: commandUuid, + state, + uuid: randomUUID(), + session_id: "test-session", + }; +} + /** The `usage` a cancelled active turn settles with when the cancel pre-empted * its result: all zeros, since only a turn's terminal result feeds the * accumulator (issue #844). */ @@ -2795,6 +2818,42 @@ describe("stop reason propagation", () => { expect(response.usage?.outputTokens).toBe(promptResult.usage.output_tokens); }); + it("ignores command_lifecycle frames without logging an unexpected-case error", async () => { + // CLIs 2.1.206+ report the fate of every uuid-stamped queued command as + // `command_lifecycle` frames (queued/started/completed/...) on the SDK + // stream, 2-3 per prompt. They are absent from the SDKMessage union, so + // without the pre-switch guard they fall through to `unreachable`'s + // error log on every prompt. + const errors: string[] = []; + const mockClient = { sessionUpdate: async () => {} } as unknown as AcpClient; + const agent = new ClaudeAcpAgent(mockClient, { + log: () => {}, + error: (msg: unknown) => errors.push(String(msg)), + }); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const { value: userMessage } = await iter.next(); + yield lifecycleFrame(userMessage.uuid, "queued"); + yield lifecycleFrame(userMessage.uuid, "started"); + yield userEcho(userMessage); + yield createResultMessage({ subtype: "success", stop_reason: "end_turn", is_error: false }); + yield lifecycleFrame(userMessage.uuid, "completed"); + yield { type: "system", subtype: "session_state_changed", state: "idle" }; + } + return messageGenerator(); + }); + + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "test" }], + }); + + expect(response.stopReason).toBe("end_turn"); + expect(errors.filter((e) => e.includes("Unexpected case"))).toEqual([]); + }); + it("settles a no-echo command result (e.g. /compact) by promoting the head turn", async () => { // Regression: /compact never echoes a user message carrying the prompt's // uuid (its only user messages are the generated summary and a @@ -6699,6 +6758,905 @@ describe("post-error recovery", () => { expect(compactResult.usage?.inputTokens).toBe(10); await agent.sessions["test-session"]?.consumer; }); + + // msg_lifecycle_v1 CLIs (2.1.206+): cancel() tracks orphans per-uuid in + // `orphanCommands`, drained by each command's own terminal lifecycle frame + // instead of by counting results. (lifecycleInit / lifecycleFrame are the + // module-scope helpers next to userEcho.) + + it("drains BOTH coalesced orphans on their one shared result (one result for two commands)", async () => { + // Two cancelled queued turns can be FOLDED into one SDK turn that emits a + // single result (observed live against CLI 2.1.206). A count would go + // stale at 1 and swallow the /compact result below, hanging its prompt — + // the shared result instead covers every "started" map entry at once + // (they were all dispatched into the emitting turn), and the trailing + // `completed` frames no-op on the already-drained entries. + const agent = createMockAgent(); + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + + const coalescedResult = createResultMessage({ + subtype: "success", + stop_reason: "end_turn", + is_error: false, + }); + coalescedResult.usage.input_tokens = 999; + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield lifecycleInit; // latch msgLifecycleV1 + yield userEcho(u1.value); // turn 1 active + const u2 = await iter.next(); // turn 2 queued (cancelled below) + const u3 = await iter.next(); // turn 3 queued (cancelled below) + await gate; // test cancels (orphans u2+u3) and sends /compact + yield { type: "system", subtype: "session_state_changed", state: "idle" }; // turn 1 settles cancelled + yield lifecycleFrame(u2.value.uuid, "started"); // both survivors dispatched + yield lifecycleFrame(u3.value.uuid, "started"); // into ONE coalesced turn + yield coalescedResult; // the shared orphan result — skipped, drains BOTH entries + yield lifecycleFrame(u2.value.uuid, "completed"); // terminal frames no-op + yield lifecycleFrame(u3.value.uuid, "completed"); // on the drained entries + await iter.next(); // /compact's pushed message — never echoes its uuid + yield { + type: "system", + subtype: "status", + status: "compacting", + session_id: "test-session", + }; + yield createResultMessage({ subtype: "success", stop_reason: "end_turn", is_error: false }); + yield { type: "system", subtype: "session_state_changed", state: "idle" }; + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn); + const second = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "second" }], + }); + const third = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "third" }], + }); + await waitFor(() => (agent.sessions["test-session"]?.turnQueue?.length ?? 0) >= 3); + + await agent.cancel({ sessionId: "test-session" }); + expect(agent.sessions["test-session"]?.orphanCommands?.size).toBe(2); + // The count lane must be untouched — the map is the only skip source here. + expect(agent.sessions["test-session"]?.pendingOrphanResults ?? 0).toBe(0); + const compact = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "/compact" }], + }); + release(); + + await expect(first).resolves.toEqual({ + stopReason: "cancelled", + usage: cancelledTurnUsage, + }); + await expect(second).resolves.toEqual({ stopReason: "cancelled" }); + await expect(third).resolves.toEqual({ stopReason: "cancelled" }); + const compactResult = await compact; + expect(compactResult.stopReason).toBe("end_turn"); + // /compact settled with its own result — the coalesced orphan's 999 was + // skipped, draining both map entries it covered. + expect(compactResult.usage?.inputTokens).toBe(10); + expect(agent.sessions["test-session"]?.orphanCommands?.size).toBe(0); + await agent.sessions["test-session"]?.consumer; + }); + + it("forgets an orphan whose `cancelled` frame precedes any `started` (dropped, no result coming)", async () => { + const agent = createMockAgent(); + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield lifecycleInit; + yield userEcho(u1.value); // turn 1 active + const u2 = await iter.next(); // turn 2 queued (cancelled below) + await gate; + yield { type: "system", subtype: "session_state_changed", state: "idle" }; // turn 1 settles cancelled + yield lifecycleFrame(u2.value.uuid, "cancelled"); // dropped before dispatch + // NO orphan result — the command never ran. + await iter.next(); // /compact's pushed message + yield { + type: "system", + subtype: "status", + status: "compacting", + session_id: "test-session", + }; + yield createResultMessage({ subtype: "success", stop_reason: "end_turn", is_error: false }); + yield { type: "system", subtype: "session_state_changed", state: "idle" }; + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn); + const second = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "second" }], + }); + await waitFor(() => (agent.sessions["test-session"]?.turnQueue?.length ?? 0) >= 2); + + await agent.cancel({ sessionId: "test-session" }); + expect(agent.sessions["test-session"]?.orphanCommands?.size).toBe(1); + const compact = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "/compact" }], + }); + release(); + + await expect(first).resolves.toEqual({ + stopReason: "cancelled", + usage: cancelledTurnUsage, + }); + await expect(second).resolves.toEqual({ stopReason: "cancelled" }); + const compactResult = await compact; + expect(compactResult.stopReason).toBe("end_turn"); + expect(compactResult.usage?.inputTokens).toBe(10); + await agent.sessions["test-session"]?.consumer; + }); + + it("skips a zombie orphan's late result exactly once (`cancelled` after `started`)", async () => { + // The orphan was dispatched, then its turn was aborted: `cancelled` after + // `started` means no more lifecycle frames come, but the dead turn's + // result may still arrive. It must be skipped (not promoted onto + // /compact) AND consume the zombie entry so it can't skip forever. + const agent = createMockAgent(); + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + + const zombieResult = createResultMessage({ + subtype: "error_during_execution", + stop_reason: null, + is_error: true, + }); + zombieResult.usage.input_tokens = 999; + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield lifecycleInit; + yield userEcho(u1.value); // turn 1 active + const u2 = await iter.next(); // turn 2 queued (cancelled below) + await gate; + yield { type: "system", subtype: "session_state_changed", state: "idle" }; // turn 1 settles cancelled + yield lifecycleFrame(u2.value.uuid, "started"); // dispatched... + yield lifecycleFrame(u2.value.uuid, "cancelled"); // ...then its turn aborted -> zombie + yield zombieResult; // the dead turn's late result — skipped, consumes the zombie + await iter.next(); // /compact's pushed message + yield { + type: "system", + subtype: "status", + status: "compacting", + session_id: "test-session", + }; + yield createResultMessage({ subtype: "success", stop_reason: "end_turn", is_error: false }); + yield { type: "system", subtype: "session_state_changed", state: "idle" }; + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn); + const second = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "second" }], + }); + await waitFor(() => (agent.sessions["test-session"]?.turnQueue?.length ?? 0) >= 2); + + await agent.cancel({ sessionId: "test-session" }); + const compact = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "/compact" }], + }); + release(); + + await expect(first).resolves.toEqual({ + stopReason: "cancelled", + usage: cancelledTurnUsage, + }); + await expect(second).resolves.toEqual({ stopReason: "cancelled" }); + const compactResult = await compact; + expect(compactResult.stopReason).toBe("end_turn"); + expect(compactResult.usage?.inputTokens).toBe(10); + expect(agent.sessions["test-session"]?.orphanCommands?.size).toBe(0); + await agent.sessions["test-session"]?.consumer; + }); + + it("forgets a pending orphan the interrupt receipt reports dropped (lifecycle lane)", async () => { + // Belt-and-suspenders: even if the CLI never emits a `cancelled` frame + // for an interrupt-dropped command, the receipt's still_queued removes + // the pending entry so /compact's echo-less result isn't swallowed. + const agent = createMockAgent(); + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield lifecycleInit; + yield userEcho(u1.value); // turn 1 active + await iter.next(); // turn 2's pushed message (dropped by the interrupt) + await gate; + yield { type: "system", subtype: "session_state_changed", state: "idle" }; // turn 1 settles cancelled + // NO lifecycle frames and NO result for turn 2 — dropped silently. + await iter.next(); // /compact's pushed message + yield { + type: "system", + subtype: "status", + status: "compacting", + session_id: "test-session", + }; + yield createResultMessage({ subtype: "success", stop_reason: "end_turn", is_error: false }); + yield { type: "system", subtype: "session_state_changed", state: "idle" }; + } + return messageGenerator(); + }); + agent.sessions["test-session"]!.query.interrupt = vi.fn(async () => ({ still_queued: [] })); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn); + // Latch the capability before cancel() picks a lane: the init frame above + // is only processed once the consumer drains it, which the first + // activation already guarantees here. + await waitFor(() => !!agent.sessions["test-session"]?.msgLifecycleV1); + const second = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "second" }], + }); + await waitFor(() => (agent.sessions["test-session"]?.turnQueue?.length ?? 0) >= 2); + + await agent.cancel({ sessionId: "test-session" }); + expect(agent.sessions["test-session"]?.orphanCommands?.size).toBe(0); + const compact = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "/compact" }], + }); + release(); + + await expect(first).resolves.toEqual({ + stopReason: "cancelled", + usage: cancelledTurnUsage, + }); + await expect(second).resolves.toEqual({ stopReason: "cancelled" }); + const compactResult = await compact; + expect(compactResult.stopReason).toBe("end_turn"); + expect(compactResult.usage?.inputTokens).toBe(10); + await agent.sessions["test-session"]?.consumer; + }); + + it("deletes (not zombifies) an orphan whose `cancelled` frame arrives AFTER its error result", async () => { + // The live-observed abort ordering is started → error result → cancelled. + // The result deletes the "started" entry when it is skipped; the late + // `cancelled` must then no-op — a zombie here would be a phantom (its + // result already came) that swallows the next echo-less result and hangs + // /compact. + const agent = createMockAgent(); + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + + const abortedResult = createResultMessage({ + subtype: "error_during_execution", + stop_reason: null, + is_error: true, + }); + abortedResult.usage.input_tokens = 999; + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield lifecycleInit; + yield userEcho(u1.value); // turn 1 active + const u2 = await iter.next(); // turn 2 queued (cancelled below) + await gate; + yield { type: "system", subtype: "session_state_changed", state: "idle" }; // turn 1 settles cancelled + yield lifecycleFrame(u2.value.uuid, "started"); // dispatched... + yield abortedResult; // ...its turn dies: error result FIRST (observed ordering) + yield lifecycleFrame(u2.value.uuid, "cancelled"); // ...terminal frame after — must no-op, not zombify + await iter.next(); // /compact's pushed message + yield { + type: "system", + subtype: "status", + status: "compacting", + session_id: "test-session", + }; + yield createResultMessage({ subtype: "success", stop_reason: "end_turn", is_error: false }); + yield { type: "system", subtype: "session_state_changed", state: "idle" }; + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn); + const second = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "second" }], + }); + await waitFor(() => (agent.sessions["test-session"]?.turnQueue?.length ?? 0) >= 2); + + await agent.cancel({ sessionId: "test-session" }); + const compact = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "/compact" }], + }); + release(); + + await expect(first).resolves.toEqual({ + stopReason: "cancelled", + usage: cancelledTurnUsage, + }); + await expect(second).resolves.toEqual({ stopReason: "cancelled" }); + const compactResult = await compact; + expect(compactResult.stopReason).toBe("end_turn"); + expect(compactResult.usage?.inputTokens).toBe(10); + expect(agent.sessions["test-session"]?.orphanCommands?.size).toBe(0); + await agent.sessions["test-session"]?.consumer; + }); + + it("drains ALL coalesced orphans of an aborted turn on its single error result", async () => { + // Two cancelled queued commands folded into ONE turn that the interrupt + // then aborts: one shared error result, then a `cancelled` frame per + // command. The result must cover BOTH entries (deleting them so the + // cancelled frames no-op) — leaving one as a zombie would swallow + // /compact's result and hang it. + const agent = createMockAgent(); + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + + const abortedResult = createResultMessage({ + subtype: "error_during_execution", + stop_reason: null, + is_error: true, + }); + abortedResult.usage.input_tokens = 999; + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield lifecycleInit; + yield userEcho(u1.value); // turn 1 active + const u2 = await iter.next(); // turn 2 queued (cancelled below) + const u3 = await iter.next(); // turn 3 queued (cancelled below) + await gate; + yield { type: "system", subtype: "session_state_changed", state: "idle" }; // turn 1 settles cancelled + yield lifecycleFrame(u2.value.uuid, "started"); // both dispatched into + yield lifecycleFrame(u3.value.uuid, "started"); // ONE coalesced turn... + yield abortedResult; // ...which aborts: ONE shared error result + yield lifecycleFrame(u2.value.uuid, "cancelled"); // per-command terminals + yield lifecycleFrame(u3.value.uuid, "cancelled"); + await iter.next(); // /compact's pushed message + yield { + type: "system", + subtype: "status", + status: "compacting", + session_id: "test-session", + }; + yield createResultMessage({ subtype: "success", stop_reason: "end_turn", is_error: false }); + yield { type: "system", subtype: "session_state_changed", state: "idle" }; + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn); + const second = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "second" }], + }); + const third = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "third" }], + }); + await waitFor(() => (agent.sessions["test-session"]?.turnQueue?.length ?? 0) >= 3); + + await agent.cancel({ sessionId: "test-session" }); + expect(agent.sessions["test-session"]?.orphanCommands?.size).toBe(2); + const compact = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "/compact" }], + }); + release(); + + await expect(first).resolves.toEqual({ + stopReason: "cancelled", + usage: cancelledTurnUsage, + }); + await expect(second).resolves.toEqual({ stopReason: "cancelled" }); + await expect(third).resolves.toEqual({ stopReason: "cancelled" }); + const compactResult = await compact; + expect(compactResult.stopReason).toBe("end_turn"); + expect(compactResult.usage?.inputTokens).toBe(10); + expect(agent.sessions["test-session"]?.orphanCommands?.size).toBe(0); + await agent.sessions["test-session"]?.consumer; + }); + + it("does not seed an orphan for a queued command whose terminal frame already passed", async () => { + // A queued command can fold into the ACTIVE turn and finish (started + + // completed frames consumed) while its Turn still sits queued. A later + // cancel() must not seed an entry for it: its one-and-only terminal frame + // is spent, so nothing would ever drain the entry and it would swallow + // /compact's echo-less result. + const agent = createMockAgent(); + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield lifecycleInit; + yield userEcho(u1.value); // turn 1 active + const u2 = await iter.next(); // turn 2 queued... + yield lifecycleFrame(u2.value.uuid, "started"); // ...folded into turn 1 + yield lifecycleFrame(u2.value.uuid, "completed"); // ...and finished pre-cancel + await gate; + yield { type: "system", subtype: "session_state_changed", state: "idle" }; // turn 1 settles cancelled + // NO further frames and NO orphan result for u2 — it is spent. + await iter.next(); // /compact's pushed message + yield { + type: "system", + subtype: "status", + status: "compacting", + session_id: "test-session", + }; + yield createResultMessage({ subtype: "success", stop_reason: "end_turn", is_error: false }); + yield { type: "system", subtype: "session_state_changed", state: "idle" }; + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn); + const second = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "second" }], + }); + // Wait for the consumer to record u2's terminal frame on the queued Turn + // before cancelling, so cancel() sees commandFinished. + await waitFor( + () => + agent.sessions["test-session"]?.turnQueue?.some( + (t: any) => t.commandFinished === "completed", + ) === true, + ); + + await agent.cancel({ sessionId: "test-session" }); + expect(agent.sessions["test-session"]?.orphanCommands?.size ?? 0).toBe(0); + const compact = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "/compact" }], + }); + release(); + + await expect(first).resolves.toEqual({ + stopReason: "cancelled", + usage: cancelledTurnUsage, + }); + await expect(second).resolves.toEqual({ stopReason: "cancelled" }); + const compactResult = await compact; + expect(compactResult.stopReason).toBe("end_turn"); + expect(compactResult.usage?.inputTokens).toBe(10); + await agent.sessions["test-session"]?.consumer; + }); + + it("stops blocking on a started orphan whose terminal frame never arrives", async () => { + // Crash caveat: a turn that dies by throwing can leak an entry with no + // terminal frame. The orphan's own result deletes the "started" entry, so + // later echo-less results must FALL THROUGH to head promotion (the + // entry's turn already produced its result) instead of being swallowed + // one after another — /compact here must settle normally with nothing + // left in the map. + const agent = createMockAgent(); + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + + const orphanResult = createResultMessage({ + subtype: "error_during_execution", + stop_reason: null, + is_error: true, + }); + orphanResult.usage.input_tokens = 999; + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield lifecycleInit; + yield userEcho(u1.value); // turn 1 active + const u2 = await iter.next(); // turn 2 queued (cancelled below) + await gate; + yield { type: "system", subtype: "session_state_changed", state: "idle" }; // turn 1 settles cancelled + yield lifecycleFrame(u2.value.uuid, "started"); // dispatched... + yield orphanResult; // ...its result arrives (entry deleted)... + // ...and its terminal frame is LOST (turn died by throwing). + await iter.next(); // /compact's pushed message + yield { + type: "system", + subtype: "status", + status: "compacting", + session_id: "test-session", + }; + yield createResultMessage({ subtype: "success", stop_reason: "end_turn", is_error: false }); + yield { type: "system", subtype: "session_state_changed", state: "idle" }; + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn); + const second = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "second" }], + }); + await waitFor(() => (agent.sessions["test-session"]?.turnQueue?.length ?? 0) >= 2); + + await agent.cancel({ sessionId: "test-session" }); + const compact = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "/compact" }], + }); + release(); + + await expect(first).resolves.toEqual({ + stopReason: "cancelled", + usage: cancelledTurnUsage, + }); + await expect(second).resolves.toEqual({ stopReason: "cancelled" }); + const compactResult = await compact; + expect(compactResult.stopReason).toBe("end_turn"); + expect(compactResult.usage?.inputTokens).toBe(10); + // The orphan's own result already deleted the entry. + expect(agent.sessions["test-session"]?.orphanCommands?.size).toBe(0); + await agent.sessions["test-session"]?.consumer; + }); + + it("consumes an orphan's result that arrives while the turn queue is empty", async () => { + // The common post-cancel timeline: the active turn settles at the + // interrupt's idle, the user hasn't typed yet, and only THEN does the + // orphaned command's dead turn flush its frames and error result. The + // bookkeeping must run even with nothing to promote — skipping it would + // leave a "started" entry that the later `cancelled` frame turns into a + // phantom zombie, swallowing the next /compact result. + const agent = createMockAgent(); + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + + const orphanResult = createResultMessage({ + subtype: "error_during_execution", + stop_reason: null, + is_error: true, + }); + orphanResult.usage.input_tokens = 999; + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield lifecycleInit; + yield userEcho(u1.value); // turn 1 active + const u2 = await iter.next(); // turn 2 queued (cancelled below) + await gate; + yield { type: "system", subtype: "session_state_changed", state: "idle" }; // turn 1 settles cancelled — queue now EMPTY + yield lifecycleFrame(u2.value.uuid, "started"); // dispatched... + yield orphanResult; // ...result arrives with an empty queue — must drain the entry + yield lifecycleFrame(u2.value.uuid, "cancelled"); // must no-op, not zombify + await iter.next(); // /compact's pushed message — only sent once the map drained + yield { + type: "system", + subtype: "status", + status: "compacting", + session_id: "test-session", + }; + yield createResultMessage({ subtype: "success", stop_reason: "end_turn", is_error: false }); + yield { type: "system", subtype: "session_state_changed", state: "idle" }; + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn); + const second = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "second" }], + }); + await waitFor(() => (agent.sessions["test-session"]?.turnQueue?.length ?? 0) >= 2); + + await agent.cancel({ sessionId: "test-session" }); + expect(agent.sessions["test-session"]?.orphanCommands?.size).toBe(1); + release(); + + await expect(first).resolves.toEqual({ + stopReason: "cancelled", + usage: cancelledTurnUsage, + }); + await expect(second).resolves.toEqual({ stopReason: "cancelled" }); + // The orphan's headless result must have drained its own entry — this + // waitFor hangs (and fails the test) if the empty-queue result skipped + // the bookkeeping. + await waitFor(() => agent.sessions["test-session"]?.orphanCommands?.size === 0); + // Only prompt /compact once the queue was provably empty at the orphan + // result, so this test pins the headless path specifically. + const compact = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "/compact" }], + }); + const compactResult = await compact; + expect(compactResult.stopReason).toBe("end_turn"); + expect(compactResult.usage?.inputTokens).toBe(10); + await agent.sessions["test-session"]?.consumer; + }); + + it("does not zombify a command folded into the active turn whose shared result was the active turn's", async () => { + // A queued command can FOLD into the still-active turn. After a cancel + // orphans it ("started" entry), the fold's shared result arrives while + // the absorbing turn is still active — attributed there, never reaching + // the echo-less skip. That result must still cover the orphan's entry; + // otherwise the trailing `cancelled` frame zombifies it into a phantom + // that swallows /compact's result. + const agent = createMockAgent(); + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + + const sharedResult = createResultMessage({ + subtype: "success", + stop_reason: "end_turn", + is_error: false, + }); + sharedResult.usage.input_tokens = 999; + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield lifecycleInit; + yield userEcho(u1.value); // turn 1 active + const u2 = await iter.next(); // turn 2 queued... + yield lifecycleFrame(u2.value.uuid, "started"); // ...folded into ACTIVE turn 1 + await gate; // test cancels (orphans u2 as "started") + yield sharedResult; // the fold's shared result — turn 1 is still active; must cover u2's entry + yield { type: "system", subtype: "session_state_changed", state: "idle" }; // turn 1 settles cancelled + yield lifecycleFrame(u2.value.uuid, "cancelled"); // must no-op, not zombify + await iter.next(); // /compact's pushed message + yield { + type: "system", + subtype: "status", + status: "compacting", + session_id: "test-session", + }; + yield createResultMessage({ subtype: "success", stop_reason: "end_turn", is_error: false }); + yield { type: "system", subtype: "session_state_changed", state: "idle" }; + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn); + const second = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "second" }], + }); + // Wait for the fold's "started" frame to latch, so cancel() seeds the + // entry as "started" (dispatched, result still coming). + await waitFor( + () => agent.sessions["test-session"]?.turnQueue?.some((t: any) => t.commandStarted) === true, + ); + + await agent.cancel({ sessionId: "test-session" }); + expect(agent.sessions["test-session"]?.orphanCommands?.size).toBe(1); + const compact = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "/compact" }], + }); + release(); + + const firstResult = await first; + expect(firstResult.stopReason).toBe("cancelled"); + // The shared result arrived pre-idle, so its usage is the cancelled + // turn's spend. + expect(firstResult.usage?.inputTokens).toBe(999); + await expect(second).resolves.toEqual({ stopReason: "cancelled" }); + const compactResult = await compact; + expect(compactResult.stopReason).toBe("end_turn"); + expect(compactResult.usage?.inputTokens).toBe(10); + expect(agent.sessions["test-session"]?.orphanCommands?.size).toBe(0); + await agent.sessions["test-session"]?.consumer; + }); + + it("does not seed a zombie for a fold-aborted command whose shared result passed BEFORE the cancel", async () => { + // The pre-cancel variant of the fold: turn 1 absorbs the queued command, + // then dies — its error result is consumed as turn 1's (rejecting it) and + // the command's `cancelled` frame latches commandFinished on the still- + // queued turn 2, all before any cancel. cancel() must then seed NOTHING + // for turn 2: its result already passed, so a zombie would be a phantom + // that swallows /compact's result. + const agent = createMockAgent(); + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + + const abortedResult = createResultMessage({ + subtype: "error_during_execution", + stop_reason: null, + is_error: true, + }); + abortedResult.usage.input_tokens = 999; + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield lifecycleInit; + yield userEcho(u1.value); // turn 1 active + const u2 = await iter.next(); // turn 2 queued... + yield lifecycleFrame(u2.value.uuid, "started"); // ...folded into ACTIVE turn 1 + yield abortedResult; // turn 1 dies: the shared error result rejects it (u2's result has now passed) + yield { type: "system", subtype: "session_state_changed", state: "idle" }; // the error's trailing idle + yield lifecycleFrame(u2.value.uuid, "cancelled"); // u2's terminal frame, consumed pre-cancel + await gate; // test cancels here — must seed nothing for u2 + await iter.next(); // /compact's pushed message + yield { + type: "system", + subtype: "status", + status: "compacting", + session_id: "test-session", + }; + yield createResultMessage({ subtype: "success", stop_reason: "end_turn", is_error: false }); + yield { type: "system", subtype: "session_state_changed", state: "idle" }; + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn); + const second = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "second" }], + }); + // turn 1's rejection confirms the shared result was consumed. + await expect(first).rejects.toThrow(); + // Wait for u2's terminal frame to latch on the queued turn before + // cancelling, so cancel() sees commandFinished === "cancelled" AND + // commandResultSeen. + await waitFor( + () => + agent.sessions["test-session"]?.turnQueue?.some( + (t: any) => t.commandFinished === "cancelled" && t.commandResultSeen === true, + ) === true, + ); + + await agent.cancel({ sessionId: "test-session" }); + expect(agent.sessions["test-session"]?.orphanCommands?.size ?? 0).toBe(0); + const compact = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "/compact" }], + }); + release(); + + await expect(second).resolves.toEqual({ stopReason: "cancelled" }); + const compactResult = await compact; + expect(compactResult.stopReason).toBe("end_turn"); + expect(compactResult.usage?.inputTokens).toBe(10); + await agent.sessions["test-session"]?.consumer; + }); + + it("does not seed an orphan on force-cancel when the wedged turn's result and terminal frame already passed", async () => { + // Wedge variant of the spent-frame rule: the interrupt aborts the active + // turn and the consumer drains its error result (dropped at the + // cancelled guard) and its `cancelled` frame — then the stream wedges + // before the trailing idle and the force-cancel backstop settles the + // turn. The backstop must seed NOTHING: both the result and the one-and- + // only terminal frame are spent, so an entry could never drain and would + // swallow /compact's result when the SDK recovers. + const agent = createMockAgent(); + agent.forceCancelGraceMs = 50; + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + let releaseWedge!: () => void; + const wedge = new Promise((resolve) => (releaseWedge = resolve)); + + const abortedResult = createResultMessage({ + subtype: "error_during_execution", + stop_reason: null, + is_error: true, + }); + abortedResult.usage.input_tokens = 999; + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield lifecycleInit; + yield userEcho(u1.value); // turn 1 active + yield lifecycleFrame(u1.value.uuid, "started"); // its own dispatch frame + await gate; // test cancels here + yield abortedResult; // dropped at the cancelled guard; marks the command's result seen + yield lifecycleFrame(u1.value.uuid, "cancelled"); // terminal frame consumed... + await wedge; // ...then the stream wedges: no trailing idle; the backstop fires + await iter.next(); // SDK "recovers": /compact's pushed message + yield { + type: "system", + subtype: "status", + status: "compacting", + session_id: "test-session", + }; + yield createResultMessage({ subtype: "success", stop_reason: "end_turn", is_error: false }); + yield { type: "system", subtype: "session_state_changed", state: "idle" }; + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn); + await waitFor( + () => agent.sessions["test-session"]?.turnQueue?.some((t: any) => t.commandStarted) === true, + ); + + await agent.cancel({ sessionId: "test-session" }); + release(); + + // The backstop settles the pending prompt after the grace elapses. + const firstResult = await first; + expect(firstResult.stopReason).toBe("cancelled"); + // Nothing seeded on either lane: result and terminal frame were spent. + expect(agent.sessions["test-session"]?.orphanCommands?.size ?? 0).toBe(0); + expect(agent.sessions["test-session"]?.pendingOrphanResults ?? 0).toBe(0); + + releaseWedge(); + const compact = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "/compact" }], + }); + const compactResult = await compact; + expect(compactResult.stopReason).toBe("end_turn"); + expect(compactResult.usage?.inputTokens).toBe(10); + await agent.sessions["test-session"]?.consumer; + }); }); describe("session/cancel wedge recovery (issue #680)", () => {