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,