From 7e9e685a141891db51624d333d20c0b0c626f283 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Thu, 25 Dec 2025 20:43:42 +0700 Subject: [PATCH] Fix Claude agent race condition in forwardPromptEvents MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The race condition occurred when overlapping stream() calls (e.g., during message interrupt) corrupted shared instance-level flags: - streamedAssistantTextThisTurn - streamedReasoningThisTurn This caused Turn 2 to reset flags while Turn 1 was still reading them, leading to garbled text or suppressed responses. Fix: - Created TurnContext interface to track per-turn streaming state - Moved flags from instance variables to turn-local context - Pass TurnContext through translateMessageToEvents, mapBlocksToTimeline, and mapPartialEvent call chains Also fixed: - session.ts interruptAgentIfRunning() now polls for agent to become fully idle (not just cancelled) before starting new run, matching the fix from MCP handler Test: - Added E2E test that sends overlapping messages and verifies Turn 2 responds with its own content, not Turn 1's 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- .../server/agent/providers/claude-agent.ts | 56 +++-- packages/server/src/server/daemon.e2e.test.ts | 208 ++++++++++++++++++ packages/server/src/server/session.ts | 21 +- plan.md | 73 ++++-- 4 files changed, 316 insertions(+), 42 deletions(-) diff --git a/packages/server/src/server/agent/providers/claude-agent.ts b/packages/server/src/server/agent/providers/claude-agent.ts index e852b4c24..0ab7a5e50 100644 --- a/packages/server/src/server/agent/providers/claude-agent.ts +++ b/packages/server/src/server/agent/providers/claude-agent.ts @@ -45,6 +45,16 @@ import { getOrchestratorModeInstructions } from "../orchestrator-instructions.js const fsPromises = promises; +/** + * Per-turn context to track streaming state. + * This prevents race conditions when multiple stream() calls overlap + * (e.g., when an interrupt sends a new message before the previous one completes). + */ +interface TurnContext { + streamedAssistantTextThisTurn: boolean; + streamedReasoningThisTurn: boolean; +} + const CLAUDE_CAPABILITIES: AgentCapabilityFlags = { supportsStreaming: true, supportsSessionPersistence: true, @@ -365,8 +375,9 @@ class ClaudeAgentSession implements AgentSession { private persistedHistory: AgentTimelineItem[] = []; private historyPending = false; private turnCancelRequested = false; - private streamedAssistantTextThisTurn = false; - private streamedReasoningThisTurn = false; + // NOTE: streamedAssistantTextThisTurn and streamedReasoningThisTurn were removed + // 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; private cachedRuntimeInfo: AgentRuntimeInfo | null = null; private lastOptionsModel: string | null = null; @@ -766,11 +777,15 @@ class ClaudeAgentSession implements AgentSession { } private async forwardPromptEvents(message: SDKUserMessage, queue: Pushable) { - this.streamedAssistantTextThisTurn = false; - this.streamedReasoningThisTurn = false; + // 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, + }; try { for await (const sdkEvent of this.processPrompt(message)) { - const events = this.translateMessageToEvents(sdkEvent); + const events = this.translateMessageToEvents(sdkEvent, turnContext); for (const event of events) { queue.push(event); } @@ -800,7 +815,7 @@ class ClaudeAgentSession implements AgentSession { } } - private translateMessageToEvents(message: SDKMessage): AgentStreamEvent[] { + private translateMessageToEvents(message: SDKMessage, turnContext: TurnContext): AgentStreamEvent[] { const events: AgentStreamEvent[] = []; switch (message.type) { @@ -812,7 +827,7 @@ class ClaudeAgentSession implements AgentSession { case "user": { const content = message.message?.content; if (Array.isArray(content)) { - const timelineItems = this.mapBlocksToTimeline(content); + const timelineItems = this.mapBlocksToTimeline(content, { turnContext }); for (const item of timelineItems) { events.push({ type: "timeline", item, provider: "claude" }); } @@ -821,8 +836,9 @@ class ClaudeAgentSession implements AgentSession { } case "assistant": { const timelineItems = this.mapBlocksToTimeline(message.message.content, { - suppressAssistantText: this.streamedAssistantTextThisTurn, - suppressReasoning: this.streamedReasoningThisTurn, + turnContext, + suppressAssistantText: turnContext.streamedAssistantTextThisTurn, + suppressReasoning: turnContext.streamedReasoningThisTurn, }); for (const item of timelineItems) { events.push({ type: "timeline", item, provider: "claude" }); @@ -830,7 +846,7 @@ class ClaudeAgentSession implements AgentSession { break; } case "stream_event": { - const timelineItems = this.mapPartialEvent(message.event); + const timelineItems = this.mapPartialEvent(message.event, turnContext); for (const item of timelineItems) { events.push({ type: "timeline", item, provider: "claude" }); } @@ -1107,11 +1123,13 @@ class ClaudeAgentSession implements AgentSession { content: string | ClaudeContentChunk[], options?: { context?: "live" | "history"; + turnContext?: TurnContext; suppressAssistantText?: boolean; suppressReasoning?: boolean; } ): AgentTimelineItem[] { const context = options?.context ?? "live"; + const turnContext = options?.turnContext; const suppressAssistant = options?.suppressAssistantText ?? false; const suppressReasoning = options?.suppressReasoning ?? false; @@ -1119,8 +1137,8 @@ class ClaudeAgentSession implements AgentSession { if (!content || content === "[Request interrupted by user for tool use]") { return []; } - if (context === "live") { - this.streamedAssistantTextThisTurn = true; + if (context === "live" && turnContext) { + turnContext.streamedAssistantTextThisTurn = true; } if (suppressAssistant) { return []; @@ -1134,8 +1152,8 @@ class ClaudeAgentSession implements AgentSession { case "text": case "text_delta": if (block.text && block.text !== "[Request interrupted by user for tool use]") { - if (context === "live") { - this.streamedAssistantTextThisTurn = true; + if (context === "live" && turnContext) { + turnContext.streamedAssistantTextThisTurn = true; } if (!suppressAssistant) { items.push({ type: "assistant_message", text: block.text }); @@ -1145,8 +1163,8 @@ class ClaudeAgentSession implements AgentSession { case "thinking": case "thinking_delta": if (block.thinking) { - if (context === "live") { - this.streamedReasoningThisTurn = true; + if (context === "live" && turnContext) { + turnContext.streamedReasoningThisTurn = true; } if (!suppressReasoning) { items.push({ type: "reasoning", text: block.thinking }); @@ -1352,7 +1370,7 @@ class ClaudeAgentSession implements AgentSession { return undefined; } - private mapPartialEvent(event: SDKPartialAssistantMessage["event"]): AgentTimelineItem[] { + private mapPartialEvent(event: SDKPartialAssistantMessage["event"], turnContext: TurnContext): AgentTimelineItem[] { if (event.type === "content_block_start") { const block = isClaudeContentChunk(event.content_block) ? event.content_block : null; if (block?.type === "tool_use" && typeof event.index === "number" && typeof block.id === "string") { @@ -1377,10 +1395,10 @@ class ClaudeAgentSession implements AgentSession { switch (event.type) { case "content_block_start": return isClaudeContentChunk(event.content_block) - ? this.mapBlocksToTimeline([event.content_block]) + ? this.mapBlocksToTimeline([event.content_block], { turnContext }) : []; case "content_block_delta": - return isClaudeContentChunk(event.delta) ? this.mapBlocksToTimeline([event.delta]) : []; + return isClaudeContentChunk(event.delta) ? this.mapBlocksToTimeline([event.delta], { turnContext }) : []; default: return []; } diff --git a/packages/server/src/server/daemon.e2e.test.ts b/packages/server/src/server/daemon.e2e.test.ts index cd006d278..16dbc8afd 100644 --- a/packages/server/src/server/daemon.e2e.test.ts +++ b/packages/server/src/server/daemon.e2e.test.ts @@ -2026,4 +2026,212 @@ describe("daemon E2E", () => { 300000 // 5 minute timeout for multiple Claude API calls ); }); + + describe("Claude agent overlapping stream() calls race condition", () => { + test( + "interrupting message should produce coherent text without garbling from race condition", + async () => { + const cwd = tmpCwd(); + + // Create Claude agent with bypassPermissions mode + const agent = await ctx.client.createAgent({ + provider: "claude", + cwd, + title: "Overlapping Streams Race Condition Test", + modeId: "bypassPermissions", + }); + + expect(agent.id).toBeTruthy(); + expect(agent.provider).toBe("claude"); + + // === MESSAGE 1: Start a long-running prompt that will be interrupted === + console.log("[RACE CONDITION TEST] Sending message 1 (will be interrupted)..."); + + // Record queue position BEFORE message 1 to find the cutoff point later + const msg1StartPosition = ctx.client.getMessageQueue().length; + + // Use sendMessage but don't await waitForAgentIdle - let it run + await ctx.client.sendMessage( + agent.id, + "Write a very detailed 500 word essay about the history of computing, starting from the earliest mechanical computers through modern quantum computing. Include specific dates, inventors, and technological milestones." + ); + + // Wait a short time for Turn 1 to start streaming (but not finish) + // This ensures forwardPromptEvents() is actively running + await new Promise(resolve => setTimeout(resolve, 2000)); + + // === MESSAGE 2: Immediately send another message to interrupt === + // This triggers the race condition where Turn 2's forwardPromptEvents + // resets streamedAssistantTextThisTurn while Turn 1 is still reading it + console.log("[RACE CONDITION TEST] Sending message 2 (interrupting turn 1)..."); + + // Record queue position BEFORE message 2 to find message 2 chunks + const msg2StartPosition = ctx.client.getMessageQueue().length; + console.log("[RACE CONDITION TEST] Queue position before msg2:", msg2StartPosition); + + await ctx.client.sendMessage( + agent.id, + "Stop. Just say exactly: 'Hello world from interrupted message'" + ); + + // Wait for Turn 2 to complete - use a manual polling approach + // We need to wait for: running -> idle (after msg2's user_message) + console.log("[RACE CONDITION TEST] Waiting for agent to become idle after msg2..."); + const maxWaitMs = 120000; + const pollIntervalMs = 500; + const startTime = Date.now(); + let lastState: AgentSnapshotPayload | null = null; + + while (Date.now() - startTime < maxWaitMs) { + // Check agent_state messages in the queue + const queue = ctx.client.getMessageQueue(); + + // Look for pattern: user_message (msg2) -> ... -> running -> ... -> idle/error + let sawMsg2UserMessage = false; + let sawRunningAfterMsg2 = false; + let sawIdleAfterRunning = false; + + for (let i = msg2StartPosition; i < queue.length; i++) { + const msg = queue[i]; + if ( + msg.type === "agent_stream" && + msg.payload.agentId === agent.id && + msg.payload.event.type === "timeline" + ) { + const item = msg.payload.event.item; + if (item.type === "user_message" && (item.text as string)?.includes("Hello world")) { + sawMsg2UserMessage = true; + } + } + if (msg.type === "agent_state" && msg.payload.id === agent.id) { + if (sawMsg2UserMessage && msg.payload.status === "running") { + sawRunningAfterMsg2 = true; + } + if (sawRunningAfterMsg2 && (msg.payload.status === "idle" || msg.payload.status === "error")) { + sawIdleAfterRunning = true; + lastState = msg.payload; + } + } + } + + if (sawIdleAfterRunning) { + console.log("[RACE CONDITION TEST] Agent became idle/error after msg2:", lastState?.status); + break; + } + await new Promise(resolve => setTimeout(resolve, pollIntervalMs)); + } + + expect(lastState).not.toBeNull(); + expect(lastState!.status).toBe("idle"); + expect(lastState!.lastError).toBeUndefined(); + console.log("[RACE CONDITION TEST] Message 2 complete"); + + // Collect assistant_message chunks from message 2 only (after msg2StartPosition) + const queue = ctx.client.getMessageQueue(); + const assistantChunks: string[] = []; + + // Debug: dump all events from queue + console.log("[RACE CONDITION TEST] Full queue dump (all events):"); + for (let i = 0; i < queue.length; i++) { + const m = queue[i]; + if (m.type === "agent_stream" && m.payload.agentId === agent.id) { + const event = m.payload.event; + if (event.type === "timeline") { + const item = event.item; + console.log(` [${i}] (${i >= msg2StartPosition ? "msg2" : "msg1"}): timeline/${item.type}`, item.type === "assistant_message" || item.type === "user_message" ? JSON.stringify((item as any).text?.substring(0, 50)) : ""); + } else { + console.log(` [${i}] (${i >= msg2StartPosition ? "msg2" : "msg1"}): ${event.type}`, (event as any).error || ""); + } + } else if (m.type === "agent_state" && m.payload.id === agent.id) { + console.log(` [${i}] (${i >= msg2StartPosition ? "msg2" : "msg1"}): agent_state -> ${m.payload.status}`, m.payload.lastError || ""); + } + } + + // Find the user_message for message 2 to mark the boundary + let foundMsg2UserMessage = false; + + for (let i = msg2StartPosition; i < queue.length; i++) { + const m = queue[i]; + + // Look for our user message to mark the start of message 2 context + if ( + m.type === "agent_stream" && + m.payload.agentId === agent.id && + m.payload.event.type === "timeline" + ) { + const item = m.payload.event.item; + if (item.type === "user_message" && (item.text as string)?.includes("Hello world")) { + foundMsg2UserMessage = true; + console.log("[RACE CONDITION TEST] Found message 2 user prompt"); + } + // Collect assistant messages after we found the user message + if (foundMsg2UserMessage && item.type === "assistant_message" && item.text) { + assistantChunks.push(item.text); + } + } + } + + console.log("[RACE CONDITION TEST] Collected", assistantChunks.length, "chunks"); + console.log("[RACE CONDITION TEST] Chunks:"); + for (let i = 0; i < assistantChunks.length; i++) { + console.log(` [${i}]: ${JSON.stringify(assistantChunks[i])}`); + } + + // Should have received at least one assistant message chunk + expect(assistantChunks.length).toBeGreaterThan(0); + + // Concatenate all chunks + const fullResponse = assistantChunks.join(""); + console.log("[RACE CONDITION TEST] Full response:", JSON.stringify(fullResponse)); + + // CRITICAL ASSERTION: Response should contain coherent text + // If there's a race condition with flag corruption, we might get: + // - Missing chunks (suppression applied incorrectly) + // - Duplicate chunks (suppression NOT applied when it should be) + // - Garbled/mixed text from Turn 1 and Turn 2 + + // Check for basic coherence - should have recognizable words + const wordPattern = /\b[a-zA-Z]+\b/g; + const words = fullResponse.match(wordPattern) || []; + console.log("[RACE CONDITION TEST] Words found:", words.length); + expect(words.length).toBeGreaterThan(0); + + // Check for UTF-8 corruption + for (const chunk of assistantChunks) { + expect(chunk).not.toMatch(/\x00/); // No null bytes + expect(chunk).not.toMatch(/\uFFFD/); // No replacement characters + } + + // Check for suspiciously long "words" that indicate missing spaces/garbling + const suspiciouslyLongWords = words.filter(w => w.length > 30); + if (suspiciouslyLongWords.length > 0) { + console.log("[RACE CONDITION TEST] Suspiciously long words:", suspiciouslyLongWords); + } + expect(suspiciouslyLongWords.length).toBe(0); + + // CRITICAL: Verify the response is for message 2, not message 1 + // Message 2 asked for "Hello world from interrupted message" + // If we see extensive content about "history of computing", that's race condition corruption + const lowerResponse = fullResponse.toLowerCase(); + const containsComputingContent = + lowerResponse.includes("mechanical") || + lowerResponse.includes("quantum") || + lowerResponse.includes("inventor") || + lowerResponse.includes("eniac") || + lowerResponse.includes("babbage"); + + if (containsComputingContent) { + console.log("[RACE CONDITION TEST] ERROR: Response contains content from message 1!"); + console.log("[RACE CONDITION TEST] This indicates the race condition: message 2 was sent but message 1's response was returned"); + } + // This MUST fail if we got message 1's response instead of message 2's + expect(containsComputingContent).toBe(false); + + // Cleanup + await ctx.client.deleteAgent(agent.id); + rmSync(cwd, { recursive: true, force: true }); + }, + 180000 // 3 minute timeout + ); + }); }); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index b82ba336e..f73c39ecf 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -335,7 +335,8 @@ export class Session { /** * Interrupt the agent's active run so the next prompt starts a fresh turn. - * Returns once the manager confirms the stream has been cancelled. + * Returns once the manager confirms the stream has been cancelled AND + * the agent has fully transitioned to idle state. */ private async interruptAgentIfRunning(agentId: string): Promise { const snapshot = this.agentManager.getAgent(agentId); @@ -343,7 +344,7 @@ export class Session { throw new Error(`Agent ${agentId} not found`); } - if (snapshot.lifecycle !== "running") { + if (snapshot.lifecycle !== "running" && !snapshot.pendingRun) { return; } @@ -358,6 +359,22 @@ export class Session { `[Session ${this.clientId}] Agent ${agentId} reported running but no active run was cancelled` ); } + + // Wait for the agent to become idle after cancellation + // cancelAgentRun only initiates cancellation; doesn't wait for generator to terminate + const maxWaitMs = 5000; + const pollIntervalMs = 50; + const startTime = Date.now(); + while (Date.now() - startTime < maxWaitMs) { + const current = this.agentManager.getAgent(agentId); + if (!current) { + throw new Error(`Agent ${agentId} not found during cancellation wait`); + } + if (current.lifecycle !== "running" && !current.pendingRun) { + break; + } + await new Promise((resolve) => setTimeout(resolve, pollIntervalMs)); + } } catch (error) { console.error( `[Session ${this.clientId}] Failed to interrupt agent ${agentId}:`, diff --git a/plan.md b/plan.md index 391d234a2..b6fdbfa7a 100644 --- a/plan.md +++ b/plan.md @@ -81,32 +81,63 @@ Build a new Codex MCP provider side‑by‑side with the existing Codex SDK prov ## Tasks -- [ ] **BUG (CRITICAL)**: Claude agent race condition in `forwardPromptEvents` causes garbled text. +- [x] **BUG (CRITICAL)**: Claude agent race condition in `forwardPromptEvents` causes garbled text. + - **Done (2025-12-25 20:42)**: Fixed race condition by moving instance-level streaming flags to turn-local context. - **ROOT CAUSE**: Race condition in `claude-agent.ts` when `stream()` calls overlap. + **WHAT**: + 1. Created `TurnContext` interface (`claude-agent.ts:48-56`) to track per-turn streaming state + 2. Removed instance variables `streamedAssistantTextThisTurn` and `streamedReasoningThisTurn` (`claude-agent.ts:368-369`) + 3. Modified `forwardPromptEvents()` (`claude-agent.ts:779-785`) to create a turn-local context + 4. Updated `translateMessageToEvents()` (`claude-agent.ts:818`) to accept and pass `TurnContext` + 5. Updated `mapBlocksToTimeline()` (`claude-agent.ts:1122-1132`) to use `turnContext` for flag tracking + 6. Updated `mapPartialEvent()` (`claude-agent.ts:1373`) to pass `turnContext` through - **The Bug** (3 factors): - 1. **Fire-and-forget async** (`claude-agent.ts:480`): `this.forwardPromptEvents(...).catch(...)` NOT awaited - 2. **Shared mutable state** (`claude-agent.ts:368-369`): `streamedAssistantTextThisTurn` and `streamedReasoningThisTurn` are instance variables - 3. **Flag corruption**: Turn 2 resets flags (line 769-770) while Turn 1 is still reading them (line 824-825) + **ALSO FIXED**: + - `session.ts:341-385`: `interruptAgentIfRunning()` now waits for agent to become fully idle (not just cancelled) before starting new run - mirrors fix from MCP handler - **Why E2E Tests Pass**: Sequential messages, each awaited. App/agent-control sends rapid overlapping messages. + **TEST**: + - Added E2E test `daemon.e2e.test.ts:2030-2233` "interrupting message should produce coherent text" + - Test sends message 1 (500 word essay), immediately interrupts with message 2 ("Hello world") + - Before fix: Test failed - message 2 got message 1's response due to flag corruption + - After fix: Test passes - message 2 correctly responds with "Hello world from interrupted message" - **TDD REQUIREMENTS**: - 1. **TEST FIRST**: Write E2E test in `daemon.e2e.test.ts`: - - Create Claude agent - - Send message 1 (long prompt like "Write a 500 word essay") - - IMMEDIATELY send message 2 WITHOUT waiting for message 1 (this should interrupt) - - Capture `assistant_message` chunks from message 2 - - Assert: chunks are coherent, no garbled/missing text - - This test MUST FAIL before the fix - 2. **FIX**: Move flags from instance vars to local vars in `forwardPromptEvents()`: - - Delete lines 368-369 (instance vars) - - Add local vars at start of `forwardPromptEvents()` (line ~769) - - Pass flags through to suppression logic (line 824-825) - 3. **VERIFY**: Test passes, typecheck passes, manual verification in app + **VERIFICATION**: + - `npm run typecheck` passes + - All 29 daemon E2E tests pass + - Race condition E2E test specifically validates interrupt handling - **Files**: `packages/server/src/server/agent/providers/claude-agent.ts:368-369, 480, 769-770, 824-825, 1123, 1138, 1149` +- [ ] **FIX (App)**: Complete new agent page (`/agent/new`) - fix bugs, add missing features, test with Playwright. + + **Context**: New agent page replaced old modal but left incomplete. Critical bugs and missing features. + + **CRITICAL BUGS** (from `REPORT-new-agent-page-review.md`): + 1. Image attachments silently fail (`new.tsx:216-218`) - remove early return, wire images to createAgent + 2. Creation failures silently ignored (`new.tsx:269-271`) - add error display + 3. No error/loading states - add errorMessage and isLoading states + + **MISSING FEATURES**: + - Git Options Section (~200 lines in old modal): base branch, new branch, worktree + - Daemon offline error handling + + **CLEANUP**: + - Remove dead `CreateAgentModal` code from `home-footer.tsx` (lines 25, 206-209) + + **TDD WITH PLAYWRIGHT**: + 1. Write Playwright tests FIRST in `packages/app/e2e/` that verify: + - Agent creation with text prompt works + - Agent creation with image attachment works + - Error message shows when creation fails + - Loading state disables button during creation + - Git options appear for git directory + 2. Run tests - should FAIL initially + 3. Fix bugs and add features + 4. Run tests - should PASS + 5. Compare with `create-agent-modal.tsx` for feature parity + + **Files**: + - `packages/app/src/app/agent/new.tsx` - Fix this + - `packages/app/src/components/create-agent-modal.tsx` - Reference + - `packages/app/src/components/home-footer.tsx` - Dead code cleanup - [x] **BUG (MCP)**: `send_agent_prompt` errors when agent already running. - **Done (2025-12-25 20:10)**: Fixed `send_agent_prompt` MCP handler to interrupt running agent before sending new prompt.