From 69b6f755fa3ad9d08fd55ba9ee48531597a9cb8e Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 28 Dec 2025 16:29:01 +0000 Subject: [PATCH] feat: add turn_canceled event for proper interruption handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a new turn_canceled event type to distinguish intentional turn interruption from errors. Fixes race conditions in Claude agent where concurrent stream() calls could cause garbled output by serializing turn handoff with turn IDs. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- packages/app/src/types/stream.ts | 1 + .../server/src/server/agent/agent-manager.ts | 13 +++ .../src/server/agent/agent-sdk-types.ts | 2 + .../agent/providers/claude-agent.test.ts | 4 + .../server/agent/providers/claude-agent.ts | 99 +++++++++++++++++-- .../server/agent/providers/codex-mcp-agent.ts | 6 +- packages/server/src/server/messages.ts | 5 + 7 files changed, 123 insertions(+), 7 deletions(-) diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index 9dca74c7d..a2b703775 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -819,6 +819,7 @@ export function reduceStreamUpdate( case "turn_started": case "turn_completed": case "turn_failed": + case "turn_canceled": case "permission_requested": case "permission_resolved": case "attention_required": diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 070209930..3a491d66a 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -403,6 +403,7 @@ export class AgentManager { const timeline: AgentTimelineItem[] = []; let finalText = ""; let usage: AgentUsage | undefined; + let canceled = false; for await (const event of events) { if (event.type === "timeline") { @@ -414,6 +415,8 @@ export class AgentManager { usage = event.usage; } else if (event.type === "turn_failed") { throw new Error(event.error); + } else if (event.type === "turn_canceled") { + canceled = true; } } @@ -423,6 +426,7 @@ export class AgentManager { finalText, usage, timeline, + canceled, }; } @@ -734,6 +738,11 @@ export class AgentManager { hasStarted = true; finish(null); } + if (event.event.type === "turn_canceled") { + currentStatus = "idle"; + hasStarted = true; + finish(null); + } } }, { agentId, replayState: true } @@ -889,6 +898,10 @@ export class AgentManager { case "turn_failed": agent.lastError = event.error; break; + case "turn_canceled": + // Cancellation is not an error, just clear any previous error + agent.lastError = undefined; + break; case "permission_requested": agent.pendingPermissions.set(event.request.id, event.request); this.emitState(agent); diff --git a/packages/server/src/server/agent/agent-sdk-types.ts b/packages/server/src/server/agent/agent-sdk-types.ts index 3fe6c816b..58f63a4e6 100644 --- a/packages/server/src/server/agent/agent-sdk-types.ts +++ b/packages/server/src/server/agent/agent-sdk-types.ts @@ -75,6 +75,7 @@ export type AgentStreamEvent = | { type: "turn_started"; provider: AgentProvider } | { type: "turn_completed"; provider: AgentProvider; usage?: AgentUsage } | { type: "turn_failed"; provider: AgentProvider; error: string } + | { type: "turn_canceled"; provider: AgentProvider; reason: string } | { type: "timeline"; item: AgentTimelineItem; provider: AgentProvider } | { type: "provider_event"; provider: AgentProvider; raw: unknown } | { type: "permission_requested"; provider: AgentProvider; request: AgentPermissionRequest } @@ -124,6 +125,7 @@ export type AgentRunResult = { finalText: string; usage?: AgentUsage; timeline: AgentTimelineItem[]; + canceled?: boolean; }; export type AgentRuntimeInfo = { diff --git a/packages/server/src/server/agent/providers/claude-agent.test.ts b/packages/server/src/server/agent/providers/claude-agent.test.ts index f0d950017..62f09f3ea 100644 --- a/packages/server/src/server/agent/providers/claude-agent.test.ts +++ b/packages/server/src/server/agent/providers/claude-agent.test.ts @@ -1256,3 +1256,7 @@ function rawContainsText(raw: unknown, text: string, depth = 0): boolean { } return false; } + +// NOTE: Turn handoff integration tests are covered by the daemon E2E test: +// "interrupting message should produce coherent text without garbling from race condition" +// in daemon.e2e.test.ts which exercises the full flow through the WebSocket API. diff --git a/packages/server/src/server/agent/providers/claude-agent.ts b/packages/server/src/server/agent/providers/claude-agent.ts index 0ab7a5e50..9351d824d 100644 --- a/packages/server/src/server/agent/providers/claude-agent.ts +++ b/packages/server/src/server/agent/providers/claude-agent.ts @@ -379,6 +379,10 @@ class ClaudeAgentSession implements AgentSession { // These flags are now tracked per-turn via TurnContext to prevent race conditions // when multiple stream() calls overlap (e.g., interrupt + new message) private cancelCurrentTurn: (() => void) | null = null; + // Track the current turn ID and active turn promise to serialize concurrent stream() calls + // and prevent race conditions where two processPrompt() loops run against the same query + private currentTurnId = 0; + private activeTurnPromise: Promise | null = null; private cachedRuntimeInfo: AgentRuntimeInfo | null = null; private lastOptionsModel: string | null = null; private managedAgentId: string | null = null; @@ -464,6 +468,35 @@ class ClaudeAgentSession implements AgentSession { prompt: AgentPromptInput, _options?: AgentRunOptions ): AsyncGenerator { + // Increment turn ID to invalidate any in-flight processPrompt() loops from previous turns. + // This prevents race conditions where an interrupted turn's events get mixed with the new turn. + const turnId = ++this.currentTurnId; + + // Actively cancel the previous turn if one exists - don't just suppress output. + // This ensures the SDK query is interrupted and side effects are stopped. + if (this.cancelCurrentTurn) { + this.cancelCurrentTurn(); + } + + // Wait briefly for any active turn to complete after cancellation. + // Use a timeout to prevent hangs if the old turn is stuck on query.next(). + if (this.activeTurnPromise) { + const TURN_HANDOFF_TIMEOUT_MS = 500; + try { + await Promise.race([ + this.activeTurnPromise, + new Promise((resolve) => setTimeout(resolve, TURN_HANDOFF_TIMEOUT_MS)), + ]); + } catch { + // Ignore errors from the previous turn - we just need it to finish or timeout + } + // Clear stale promise to prevent accumulation of stuck turns + this.activeTurnPromise = null; + } + + // Reset cancel flag at the start of each turn to prevent stale state from previous turns + this.turnCancelRequested = false; + const sdkMessage = this.toSdkUserMessage(prompt); const queue = new Pushable(); this.eventQueue = queue; @@ -478,6 +511,12 @@ class ClaudeAgentSession implements AgentSession { this.interruptActiveTurn().catch((error) => { console.warn("[ClaudeAgentSession] Failed to interrupt during cancel:", error); }); + // Push turn_canceled before ending the queue so consumers get proper lifecycle signals + queue.push({ + type: "turn_canceled", + provider: "claude", + reason: "Interrupted", + }); queue.end(); }; this.cancelCurrentTurn = requestCancel; @@ -488,14 +527,22 @@ class ClaudeAgentSession implements AgentSession { this.historyPending = false; this.persistedHistory = []; } - this.forwardPromptEvents(sdkMessage, queue).catch((error) => { + + // Start forwarding events and track the promise so future turns can wait for completion + const forwardPromise = this.forwardPromptEvents(sdkMessage, queue, turnId); + this.activeTurnPromise = forwardPromise; + forwardPromise.catch((error) => { console.error("[ClaudeAgentSession] Unexpected error in forwardPromptEvents:", error); }); try { for await (const event of queue) { yield event; - if (event.type === "turn_completed" || event.type === "turn_failed") { + if ( + event.type === "turn_completed" || + event.type === "turn_failed" || + event.type === "turn_canceled" + ) { finishedNaturally = true; break; } @@ -510,6 +557,10 @@ class ClaudeAgentSession implements AgentSession { if (this.cancelCurrentTurn === requestCancel) { this.cancelCurrentTurn = null; } + // Clear the active turn promise if it's still ours + if (this.activeTurnPromise === forwardPromise) { + this.activeTurnPromise = null; + } } } @@ -752,7 +803,8 @@ class ClaudeAgentSession implements AgentSession { } private async *processPrompt( - sdkMessage: SDKUserMessage + sdkMessage: SDKUserMessage, + turnId: number ): AsyncGenerator { const query = await this.ensureQuery(); if (!this.input) { @@ -762,6 +814,12 @@ class ClaudeAgentSession implements AgentSession { this.input.push(sdkMessage); while (true) { + // Check if this turn has been superseded by a new one. + // This prevents stale events from an interrupted turn from being yielded. + if (this.currentTurnId !== turnId) { + break; + } + const { value, done } = await query.next(); if (done) { break; @@ -769,6 +827,12 @@ class ClaudeAgentSession implements AgentSession { if (!value) { continue; } + + // Double-check turn ID after awaiting, in case a new turn started while we waited + if (this.currentTurnId !== turnId) { + break; + } + yield value; if (value.type === "result") { break; @@ -776,22 +840,34 @@ class ClaudeAgentSession implements AgentSession { } } - private async forwardPromptEvents(message: SDKUserMessage, queue: Pushable) { + private async forwardPromptEvents( + message: SDKUserMessage, + queue: Pushable, + turnId: number + ) { // Create a turn-local context to track streaming state. // This prevents race conditions when a new stream() call interrupts a running one. const turnContext: TurnContext = { streamedAssistantTextThisTurn: false, streamedReasoningThisTurn: false, }; + let completedNormally = false; try { - for await (const sdkEvent of this.processPrompt(message)) { + for await (const sdkEvent of this.processPrompt(message, turnId)) { + // Check if this turn has been superseded before pushing events + if (this.currentTurnId !== turnId) { + break; + } const events = this.translateMessageToEvents(sdkEvent, turnContext); for (const event of events) { queue.push(event); + if (event.type === "turn_completed") { + completedNormally = true; + } } } } catch (error) { - if (!this.turnCancelRequested) { + if (!this.turnCancelRequested && this.currentTurnId === turnId) { queue.push({ type: "turn_failed", provider: "claude", @@ -799,6 +875,17 @@ class ClaudeAgentSession implements AgentSession { }); } } finally { + // Emit terminal event for superseded turns so consumers get proper lifecycle signals. + // Use turn_canceled (not turn_failed) to distinguish intentional interruption from errors. + // Only emit if not already emitted by requestCancel() (indicated by turnCancelRequested). + const wasSuperseded = this.currentTurnId !== turnId; + if (wasSuperseded && !completedNormally && !this.turnCancelRequested) { + queue.push({ + type: "turn_canceled", + provider: "claude", + reason: "Interrupted by new message", + }); + } this.turnCancelRequested = false; queue.end(); } diff --git a/packages/server/src/server/agent/providers/codex-mcp-agent.ts b/packages/server/src/server/agent/providers/codex-mcp-agent.ts index 998e4473e..5c54a1e2c 100644 --- a/packages/server/src/server/agent/providers/codex-mcp-agent.ts +++ b/packages/server/src/server/agent/providers/codex-mcp-agent.ts @@ -2856,7 +2856,11 @@ class CodexMcpAgentSession implements AgentSession { try { for await (const event of queue) { yield event; - if (event.type === "turn_completed" || event.type === "turn_failed") { + if ( + event.type === "turn_completed" || + event.type === "turn_failed" || + event.type === "turn_canceled" + ) { break; } } diff --git a/packages/server/src/server/messages.ts b/packages/server/src/server/messages.ts index bbca85fd9..19c7b113e 100644 --- a/packages/server/src/server/messages.ts +++ b/packages/server/src/server/messages.ts @@ -187,6 +187,11 @@ export const AgentStreamEventPayloadSchema = z.discriminatedUnion("type", [ provider: AgentProviderSchema, error: z.string(), }), + z.object({ + type: z.literal("turn_canceled"), + provider: AgentProviderSchema, + reason: z.string(), + }), z.object({ type: z.literal("timeline"), provider: AgentProviderSchema,