mirror of
https://github.com/agentclientprotocol/claude-agent-acp.git
synced 2026-10-27 04:30:59 +00:00
fix: Handle SDK idle turns without results (#835)
Closes #825 (hopefully)
This commit is contained in:
181
src/acp-agent.ts
181
src/acp-agent.ts
@@ -176,6 +176,14 @@ const DEFAULT_CONTEXT_WINDOW = 200000;
|
||||
* pre-empt a slow-but-healthy interrupt. */
|
||||
const DEFAULT_FORCE_CANCEL_GRACE_MS = 30_000;
|
||||
|
||||
/** Error surfaced when the SDK declares a turn over (`session_state_changed:
|
||||
* idle`, its authoritative turn-over signal) without ever emitting the turn's
|
||||
* `result` — a model stream that dropped mid-turn, or an async agent that
|
||||
* completed/stalled without the host turn resolving (issue #825). */
|
||||
const TURN_NO_RESULT_MESSAGE =
|
||||
"The turn ended without a result: the agent went idle while this prompt was still in flight " +
|
||||
"(e.g. the model stream dropped mid-turn). Any partial output may be incomplete; please retry.";
|
||||
|
||||
/** Internal model-selection state. Mirrors the shape the ACP SDK exposed as
|
||||
* `SessionModelState` before model selection moved entirely into
|
||||
* `SessionConfigOption` (category "model"). Retained internally to track the
|
||||
@@ -1149,6 +1157,20 @@ 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";
|
||||
// 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;
|
||||
|
||||
const resetTurnScratch = () => {
|
||||
lastAssistantTotalUsage = null;
|
||||
@@ -1277,9 +1299,21 @@ export class ClaudeAcpAgent {
|
||||
// after each fire so the consumer keeps serving later turns.
|
||||
let cancelController = session.cancelController!;
|
||||
|
||||
// The in-flight query.next(), kept across abort wake-ups that don't
|
||||
// consume a message, so no yielded message is ever dropped — async
|
||||
// generators serialize next() calls, so racing a SECOND next() while one
|
||||
// is pending would make the abandoned one swallow a message (e.g. a
|
||||
// force-cancelled turn's late result, whose orphan accounting below
|
||||
// depends on actually seeing it).
|
||||
let pendingNext: Promise<{ kind: "message"; result: IteratorResult<SDKMessage, void> }> | null =
|
||||
null;
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
const nextMessage = session.query.next();
|
||||
pendingNext ??= session.query
|
||||
.next()
|
||||
.then((result) => ({ kind: "message" as const, result }));
|
||||
const nextMessage = pendingNext;
|
||||
// Fresh abort listener per iteration, removed when next() wins, so a
|
||||
// long-lived session doesn't accumulate listeners on one signal.
|
||||
let onAbort!: () => void;
|
||||
@@ -1287,20 +1321,29 @@ export class ClaudeAcpAgent {
|
||||
onAbort = () => resolve("abort");
|
||||
cancelController.signal.addEventListener("abort", onAbort, { once: true });
|
||||
});
|
||||
const raced = await Promise.race([
|
||||
nextMessage.then((result) => ({ kind: "message" as const, result })),
|
||||
abortRace,
|
||||
]);
|
||||
const raced = await Promise.race([nextMessage, abortRace]);
|
||||
cancelController.signal.removeEventListener("abort", onAbort);
|
||||
|
||||
if (raced === "abort") {
|
||||
// cancel()/teardown woke us. Abandon the in-flight next() (swallowing
|
||||
// any later rejection so it can't surface as unhandled) and settle the
|
||||
// active turn "cancelled" per the ACP contract. If the session is
|
||||
// being torn down, stop; otherwise re-arm and keep consuming.
|
||||
void nextMessage.catch(() => {});
|
||||
// cancel()/teardown woke us: settle the active turn "cancelled" per
|
||||
// the ACP contract. The SDK never acknowledged this turn (that's why
|
||||
// the force-cancel backstop fired), so if it later recovers from the
|
||||
// wedge it will still emit the turn's result — with no live turn to
|
||||
// match — followed by its trailing idle. Pre-count it as an orphan
|
||||
// so that late result is skipped (not promoted onto the next queued
|
||||
// prompt) and its trailer is recorded as owed, not read as the next
|
||||
// turn being abandoned. Stale counts self-heal: activation resets
|
||||
// them (see activateTurn).
|
||||
if (session.activeTurn && !session.activeTurn.settled) {
|
||||
session.pendingOrphanResults = (session.pendingOrphanResults ?? 0) + 1;
|
||||
}
|
||||
settleActive({ stopReason: "cancelled" });
|
||||
// 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`
|
||||
// stays in flight so its eventual message is processed, not dropped.
|
||||
if (!this.sessions[params.sessionId]) {
|
||||
void nextMessage.catch(() => {});
|
||||
return;
|
||||
}
|
||||
cancelController = new AbortController();
|
||||
@@ -1308,6 +1351,9 @@ export class ClaudeAcpAgent {
|
||||
continue;
|
||||
}
|
||||
|
||||
// A message arrived: this next() is consumed; arm a fresh one next pass.
|
||||
pendingNext = null;
|
||||
|
||||
const { value: message, done } = raced.result as IteratorResult<SDKMessage, void>;
|
||||
|
||||
if (done || !message) {
|
||||
@@ -1445,19 +1491,61 @@ export class ClaudeAcpAgent {
|
||||
}
|
||||
case "session_state_changed": {
|
||||
if (message.state === "idle") {
|
||||
// A non-cancelled turn already settled at its terminal
|
||||
// `result` (issue #773), so this trailing `idle` is just
|
||||
// absorbed. We must NOT settle `activeTurn` here in that case:
|
||||
// `idle` carries no turn identity, and it can lag (the SDK
|
||||
// flushes held-back results / drains background agents first),
|
||||
// so by the time it arrives the SDK may have echoed the NEXT
|
||||
// turn and activated it — settling now would resolve that new
|
||||
// turn prematurely with end_turn and ~zero usage, dropping its
|
||||
// real result. Only a cancelled turn relies on `idle`: its
|
||||
// `result` is dropped at the `session.cancelled` guard, so it
|
||||
// never settles at a result and must settle here.
|
||||
if (session.cancelled) {
|
||||
// A non-cancelled turn normally settled at its terminal
|
||||
// `result` already (issue #773), and that result recorded an
|
||||
// owed trailing idle — absorbed here via the decrement. We
|
||||
// must NOT settle `activeTurn` on an owed idle: `idle`
|
||||
// carries no turn identity, and it can lag (the SDK flushes
|
||||
// held-back results / drains background agents first), so by
|
||||
// the time it arrives the SDK may have echoed the NEXT turn
|
||||
// and activated it — settling now would resolve that new
|
||||
// turn prematurely with end_turn and ~zero usage, dropping
|
||||
// its real result. A cancelled turn relies on `idle`: its
|
||||
// `result` is dropped at the `session.cancelled` guard, so
|
||||
// it never settles at a result and must settle here.
|
||||
//
|
||||
// An idle that is NOT owed while the active turn is still
|
||||
// unsettled is the issue #825 signature: `idle` is the SDK's
|
||||
// authoritative turn-over signal (it fires after held-back
|
||||
// results flush and background agents drain), so a turn that
|
||||
// reaches it without a result will never get one — the model
|
||||
// stream dropped mid-turn, or an async agent
|
||||
// completed/stalled without the host turn resolving. Fail
|
||||
// the turn NOW so its session/prompt gets a terminal
|
||||
// response, instead of leaving it hanging until the next
|
||||
// prompt drains the wreckage.
|
||||
if (session.cancelled && session.activeTurn && !session.activeTurn.settled) {
|
||||
settleActive({ stopReason: "cancelled" });
|
||||
} else if (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--;
|
||||
} else if (
|
||||
!session.cancelled &&
|
||||
session.activeTurn &&
|
||||
!session.activeTurn.settled
|
||||
) {
|
||||
// Deliberately only the ACTIVE turn: a queued turn that
|
||||
// was never echoed is NOT failed here, because an idle
|
||||
// can legitimately precede the SDK picking up freshly
|
||||
// pushed input (the idle was emitted before the SDK read
|
||||
// it) — failing the queue head on that race would reject
|
||||
// a prompt the SDK is about to run. A turn abandoned
|
||||
// before its echo therefore still hangs until cancel or
|
||||
// the next prompt; only a timer could tell those apart.
|
||||
this.logger.error(
|
||||
`Session ${params.sessionId}: SDK went idle without emitting a result ` +
|
||||
`for the active turn; failing the in-flight prompt (issue #825)`,
|
||||
);
|
||||
failActive(
|
||||
RequestError.internalError(
|
||||
errorKindData("no_result"),
|
||||
TURN_NO_RESULT_MESSAGE,
|
||||
),
|
||||
);
|
||||
}
|
||||
// The SDK generates the session title in a background task and
|
||||
// persists it to the session file; `idle` is the turn-over
|
||||
@@ -1710,6 +1798,24 @@ export class ClaudeAcpAgent {
|
||||
ensureActiveTurn();
|
||||
}
|
||||
|
||||
// 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
|
||||
// handler absorbs that idle rather than reading it as a turn the
|
||||
// SDK abandoned (issue #825). One exclusion: the cancelled ACTIVE
|
||||
// turn's own result. It is dropped at the `session.cancelled`
|
||||
// guard, and either the idle itself settles the turn (consuming
|
||||
// the trailer) or the next echo's hand-off does (which records
|
||||
// the debt there instead) — counting here too would double it.
|
||||
// Results skipped while cancelled with NO active turn — orphaned
|
||||
// queued turns the SDK still ran, or a force-cancelled turn's
|
||||
// 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++;
|
||||
}
|
||||
|
||||
// Accumulate usage into the user turn's tally. Skip task-notification
|
||||
// followups: their cost is real but is reported separately via the
|
||||
// usage_update below, and `session.accumulatedUsage` is only reset on
|
||||
@@ -2026,11 +2132,17 @@ export class ClaudeAcpAgent {
|
||||
// "cancelled" per the ACP contract rather than "end_turn" —
|
||||
// otherwise a cancel followed quickly by the next prompt
|
||||
// would report the cancelled turn as a normal completion.
|
||||
settleActive(
|
||||
session.cancelled
|
||||
? { stopReason: "cancelled" }
|
||||
: { stopReason: "end_turn", usage: sessionUsage(session) },
|
||||
);
|
||||
if (session.cancelled) {
|
||||
// The cancelled turn settles here, but the trailing idle
|
||||
// its interrupt produces is still in flight — record the
|
||||
// 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++;
|
||||
settleActive({ stopReason: "cancelled" });
|
||||
} else {
|
||||
settleActive({ stopReason: "end_turn", usage: sessionUsage(session) });
|
||||
}
|
||||
}
|
||||
activateTurn(queued);
|
||||
}
|
||||
@@ -3842,19 +3954,24 @@ function totalTokens(usage: UsageSnapshot): number {
|
||||
);
|
||||
}
|
||||
|
||||
/** Error kinds this adapter invents itself, alongside the SDK's categorical
|
||||
* `SDKAssistantMessageError` kinds: `no_result` marks a turn the SDK declared
|
||||
* over without ever emitting its result (issue #825). */
|
||||
type AgentErrorKind = SDKAssistantMessageError | "no_result";
|
||||
|
||||
/**
|
||||
* Build the `data` payload attached to a `RequestError.internalError` when we
|
||||
* have a categorical error from the Claude SDK. Returns `undefined` when no
|
||||
* categorical error is available, matching the previous behavior of passing
|
||||
* `undefined` to `RequestError.internalError`.
|
||||
* have a categorical error — from the Claude SDK, or one of the adapter's own
|
||||
* kinds. Returns `undefined` when no categorical error is available, matching
|
||||
* the previous behavior of passing `undefined` to `RequestError.internalError`.
|
||||
*
|
||||
* The `errorKind` field is a convention for ACP clients to dispatch on
|
||||
* without having to pattern-match the human-readable message text. Clients
|
||||
* that don't understand it fall back to the existing message-based rendering.
|
||||
*/
|
||||
function errorKindData(
|
||||
errorKind: SDKAssistantMessageError | undefined,
|
||||
): { errorKind: SDKAssistantMessageError } | undefined {
|
||||
errorKind: AgentErrorKind | undefined,
|
||||
): { errorKind: AgentErrorKind } | undefined {
|
||||
return errorKind ? { errorKind } : undefined;
|
||||
}
|
||||
|
||||
|
||||
@@ -4425,9 +4425,11 @@ describe("usage_update computation", () => {
|
||||
|
||||
it("compact_boundary uses authoritative getContextUsage for used, keeps session window for size", async () => {
|
||||
const { agent, updates } = createMockAgentWithCapture();
|
||||
// No trailing idle: an idle with no preceding result now fails the turn as
|
||||
// abandoned (issue #825), and a real compaction turn always produces a
|
||||
// result. Here the stream simply ends, settling the prompt end_turn.
|
||||
injectSession(agent, [
|
||||
{ type: "system", subtype: "compact_boundary", session_id: "test-session" },
|
||||
{ type: "system", subtype: "session_state_changed", state: "idle" },
|
||||
]);
|
||||
const session = agent.sessions["test-session"];
|
||||
// A 1M window learned earlier (e.g. from modelUsage) must survive compaction
|
||||
@@ -4449,9 +4451,9 @@ describe("usage_update computation", () => {
|
||||
|
||||
it("compact_boundary falls back to used:0 when getContextUsage fails", async () => {
|
||||
const { agent, updates } = createMockAgentWithCapture();
|
||||
// No trailing idle — see the sibling test above (issue #825).
|
||||
injectSession(agent, [
|
||||
{ type: "system", subtype: "compact_boundary", session_id: "test-session" },
|
||||
{ type: "system", subtype: "session_state_changed", state: "idle" },
|
||||
]);
|
||||
const session = agent.sessions["test-session"];
|
||||
session.contextWindowSize = 200000;
|
||||
@@ -6251,6 +6253,264 @@ describe("session/cancel wedge recovery (issue #680)", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("turn abandoned by the SDK (issue #825)", () => {
|
||||
function createMockAgent() {
|
||||
const mockClient = {
|
||||
sessionUpdate: async () => {},
|
||||
} as unknown as AcpClient;
|
||||
return new ClaudeAcpAgent(mockClient, { log: () => {}, error: () => {} });
|
||||
}
|
||||
|
||||
function createResultMessage() {
|
||||
return {
|
||||
type: "result" as const,
|
||||
subtype: "success",
|
||||
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",
|
||||
};
|
||||
}
|
||||
|
||||
// Deterministic wait for the consumer to reach a state (e.g. a turn became
|
||||
// active), instead of a timing-based sleep that can pass vacuously on a
|
||||
// slow machine while exercising the wrong code path.
|
||||
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");
|
||||
};
|
||||
|
||||
it("fails the in-flight prompt when the SDK goes idle without emitting a result", async () => {
|
||||
// The issue #825 signature: the model stream drops mid-turn, the SDK's
|
||||
// turn loop exits (trailing `session_state_changed: idle` — its
|
||||
// authoritative turn-over signal) but the turn's `result` never arrives.
|
||||
// The prompt must fail at that idle instead of hanging until the next
|
||||
// prompt drains the stale state.
|
||||
const agent = createMockAgent();
|
||||
injectGeneratorSession(agent, (input) => {
|
||||
async function* messageGenerator() {
|
||||
const iter = input[Symbol.asyncIterator]();
|
||||
const u1 = await iter.next();
|
||||
yield userEcho(u1.value); // turn 1 active
|
||||
yield { type: "system", subtype: "session_state_changed", state: "idle" };
|
||||
// Parked awaiting the next prompt's input, like the real wedged SDK.
|
||||
const u2 = await iter.next();
|
||||
yield userEcho(u2.value);
|
||||
yield createResultMessage();
|
||||
yield { type: "system", subtype: "session_state_changed", state: "idle" };
|
||||
}
|
||||
return messageGenerator();
|
||||
});
|
||||
|
||||
await expect(
|
||||
agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "first" }] }),
|
||||
).rejects.toThrow(/without a result/);
|
||||
|
||||
// The session recovers: the next prompt runs normally on the same stream.
|
||||
const second = await agent.prompt({
|
||||
sessionId: "test-session",
|
||||
prompt: [{ type: "text", text: "second" }],
|
||||
});
|
||||
expect(second.stopReason).toBe("end_turn");
|
||||
expect(second.usage?.inputTokens).toBe(10);
|
||||
});
|
||||
|
||||
it("absorbs a cancelled turn's lagged trailing idle without failing the next turn", async () => {
|
||||
// Cancel turn 1; turn 2's echo arrives BEFORE turn 1's trailing idle, so
|
||||
// the hand-off settles turn 1 "cancelled" (recording the owed idle). The
|
||||
// lagged idle then lands while healthy turn 2 is active — it must be
|
||||
// absorbed as turn 1's trailer, not read as turn 2 ending without a
|
||||
// result (which would reject turn 2 here).
|
||||
const agent = createMockAgent();
|
||||
agent.forceCancelGraceMs = 60_000; // hand-off must settle turn 1, not the backstop
|
||||
injectGeneratorSession(agent, (input) => {
|
||||
async function* messageGenerator() {
|
||||
const iter = input[Symbol.asyncIterator]();
|
||||
const u1 = await iter.next();
|
||||
yield userEcho(u1.value); // turn 1 active
|
||||
// Parks until the test (after cancelling turn 1) sends turn 2.
|
||||
const u2 = await iter.next();
|
||||
yield userEcho(u2.value); // hand-off: turn 1 settles cancelled
|
||||
yield { type: "system", subtype: "session_state_changed", state: "idle" }; // turn 1's lagged trailer
|
||||
yield createResultMessage();
|
||||
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 agent.cancel({ sessionId: "test-session" });
|
||||
const second = agent.prompt({
|
||||
sessionId: "test-session",
|
||||
prompt: [{ type: "text", text: "second" }],
|
||||
});
|
||||
|
||||
await expect(first).resolves.toEqual({ stopReason: "cancelled" });
|
||||
const secondResult = await second;
|
||||
expect(secondResult.stopReason).toBe("end_turn");
|
||||
expect(secondResult.usage?.inputTokens).toBe(10);
|
||||
});
|
||||
|
||||
it("counts an orphaned queued turn's trailer so it cannot false-fail a later prompt", async () => {
|
||||
// Turn 1 active, turn 2 queued. cancel() settles+removes turn 2, but its
|
||||
// message was already pushed, so the SDK still runs it and emits a result
|
||||
// (the orphan) plus a trailing idle. The orphan result arrives while
|
||||
// session.cancelled is still true and must record the owed trailer —
|
||||
// otherwise the orphan's idle, lagging past turn 3's echo, would be read
|
||||
// as turn 3 ending without a result and reject the healthy prompt.
|
||||
const agent = createMockAgent();
|
||||
let releaseAfterCancel!: () => void;
|
||||
const afterCancel = new Promise<void>((resolve) => (releaseAfterCancel = resolve));
|
||||
injectGeneratorSession(agent, (input) => {
|
||||
async function* messageGenerator() {
|
||||
const iter = input[Symbol.asyncIterator]();
|
||||
const u1 = await iter.next();
|
||||
yield userEcho(u1.value); // turn 1 active
|
||||
await iter.next(); // turn 2's pushed message (cancelled + removed)
|
||||
await afterCancel; // test cancels (orphaning turn 2)
|
||||
yield { type: "system", subtype: "session_state_changed", state: "idle" }; // settles turn 1 cancelled
|
||||
yield createResultMessage(); // turn 2's orphan result — skipped; trailer now owed
|
||||
const u3 = await iter.next();
|
||||
yield userEcho(u3.value); // turn 3 activates (clears cancelled)
|
||||
yield { type: "system", subtype: "session_state_changed", state: "idle" }; // orphan's lagged trailer — absorbed
|
||||
yield createResultMessage();
|
||||
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 agent.cancel({ sessionId: "test-session" });
|
||||
releaseAfterCancel();
|
||||
|
||||
await expect(first).resolves.toEqual({ stopReason: "cancelled" });
|
||||
await expect(second).resolves.toEqual({ stopReason: "cancelled" });
|
||||
const third = await agent.prompt({
|
||||
sessionId: "test-session",
|
||||
prompt: [{ type: "text", text: "third" }],
|
||||
});
|
||||
expect(third.stopReason).toBe("end_turn");
|
||||
expect(third.usage?.inputTokens).toBe(10);
|
||||
});
|
||||
|
||||
it("skips a force-cancelled turn's late result and absorbs its trailer after recovery", async () => {
|
||||
// A wedged turn is settled "cancelled" by the force-cancel backstop; the
|
||||
// SDK later recovers from the wedge and still emits that turn's result
|
||||
// and trailing idle. The late result must be skipped as an orphan — not
|
||||
// promoted onto the next queued prompt (which would settle it with the
|
||||
// stale turn's stop reason and usage) — and its trailer absorbed, not
|
||||
// read as the next turn being abandoned.
|
||||
const agent = createMockAgent();
|
||||
agent.forceCancelGraceMs = 10;
|
||||
let releaseRecovery!: () => void;
|
||||
const recovery = new Promise<void>((resolve) => (releaseRecovery = resolve));
|
||||
const staleResult = createResultMessage();
|
||||
staleResult.usage.input_tokens = 999;
|
||||
injectGeneratorSession(agent, (input) => {
|
||||
async function* messageGenerator() {
|
||||
const iter = input[Symbol.asyncIterator]();
|
||||
const u1 = await iter.next();
|
||||
yield userEcho(u1.value); // turn 1 active
|
||||
await recovery; // wedged: interrupt is a no-op; the backstop settles turn 1
|
||||
yield staleResult; // turn 1's late result — orphan-skipped; trailer now owed
|
||||
const u2 = await iter.next();
|
||||
yield userEcho(u2.value); // turn 2 activates (clears cancelled)
|
||||
yield { type: "system", subtype: "session_state_changed", state: "idle" }; // turn 1's lagged trailer — absorbed
|
||||
yield createResultMessage();
|
||||
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 agent.cancel({ sessionId: "test-session" }); // backstop (10ms) settles turn 1
|
||||
await expect(first).resolves.toEqual({ stopReason: "cancelled" });
|
||||
|
||||
// Queue turn 2 BEFORE the SDK recovers, so the stale result races it.
|
||||
const second = agent.prompt({
|
||||
sessionId: "test-session",
|
||||
prompt: [{ type: "text", text: "second" }],
|
||||
});
|
||||
releaseRecovery();
|
||||
|
||||
const secondResult = await second;
|
||||
expect(secondResult.stopReason).toBe("end_turn");
|
||||
// Turn 2 settles with its OWN result's usage — not the stale 999.
|
||||
expect(secondResult.usage?.inputTokens).toBe(10);
|
||||
});
|
||||
|
||||
it("does not leak the owed idle when a cancel lands between a result and its trailer", async () => {
|
||||
// Turn 1's result settles it (debt recorded); the user cancels before the
|
||||
// lagged trailing idle arrives. That idle must still be absorbed via the
|
||||
// debt — otherwise it leaks and a future genuine abandoned-turn idle
|
||||
// would be absorbed instead of detected. Turn 2 then wedges (idle, no
|
||||
// result) and must still be failed.
|
||||
const agent = createMockAgent();
|
||||
let releaseAfterCancel!: () => void;
|
||||
const afterCancel = new Promise<void>((resolve) => (releaseAfterCancel = resolve));
|
||||
injectGeneratorSession(agent, (input) => {
|
||||
async function* messageGenerator() {
|
||||
const iter = input[Symbol.asyncIterator]();
|
||||
const u1 = await iter.next();
|
||||
yield userEcho(u1.value);
|
||||
yield createResultMessage(); // turn 1 settles here; trailer now owed
|
||||
await afterCancel; // test cancels with no active turn
|
||||
yield { type: "system", subtype: "session_state_changed", state: "idle" }; // absorbed via debt
|
||||
const u2 = await iter.next();
|
||||
yield userEcho(u2.value); // turn 2 active
|
||||
yield { type: "system", subtype: "session_state_changed", state: "idle" }; // un-owed → turn 2 abandoned
|
||||
}
|
||||
return messageGenerator();
|
||||
});
|
||||
|
||||
const first = await agent.prompt({
|
||||
sessionId: "test-session",
|
||||
prompt: [{ type: "text", text: "first" }],
|
||||
});
|
||||
expect(first.stopReason).toBe("end_turn");
|
||||
|
||||
await agent.cancel({ sessionId: "test-session" });
|
||||
releaseAfterCancel();
|
||||
|
||||
await expect(
|
||||
agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "second" }] }),
|
||||
).rejects.toThrow(/without a result/);
|
||||
});
|
||||
});
|
||||
|
||||
describe("streamEventToAcpNotifications", () => {
|
||||
it("treats `ping` keep-alive events as no-ops without logging to stderr", () => {
|
||||
const errors: unknown[][] = [];
|
||||
|
||||
Reference in New Issue
Block a user