diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index e7e8788f9..3b33f8b3d 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -545,15 +545,9 @@ export class AgentManager { } try { - const cancellation = pendingRun.return( - undefined as unknown as AgentStreamEvent - ); - void cancellation.catch((error) => { - console.error( - `[AgentManager] Failed to cancel run for agent ${agentId}:`, - error - ); - }); + // Await the generator's .return() to ensure the finally block runs + // and pendingRun is properly cleared before we return. + await pendingRun.return(undefined as unknown as AgentStreamEvent); return true; } catch (error) { console.error( diff --git a/packages/server/src/server/agent/providers/claude-agent.ts b/packages/server/src/server/agent/providers/claude-agent.ts index a93540124..f5e773d7f 100644 --- a/packages/server/src/server/agent/providers/claude-agent.ts +++ b/packages/server/src/server/agent/providers/claude-agent.ts @@ -378,6 +378,9 @@ 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 pending interrupt promise so we can await it in processPrompt + // This ensures the interrupt's response is consumed before we call query.next() + private pendingInterruptPromise: Promise | 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; @@ -481,28 +484,12 @@ class ClaudeAgentSession implements AgentSession { // 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. + // Cancel the previous turn if one exists. The caller of interrupt() is responsible + // for awaiting completion - the new turn just signals cancellation and proceeds. 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; @@ -517,7 +504,8 @@ class ClaudeAgentSession implements AgentSession { } cancelIssued = true; this.turnCancelRequested = true; - this.interruptActiveTurn().catch((error) => { + // Store the interrupt promise so processPrompt can await it before calling query.next() + this.pendingInterruptPromise = 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 @@ -734,6 +722,9 @@ class ClaudeAgentSession implements AgentSession { MCP_TIMEOUT: "600000", MCP_TOOL_TIMEOUT: "600000", }, + // If we have a session ID from a previous query (e.g., after interrupt), + // resume that session to continue the conversation history. + ...(this.claudeSessionId ? { resume: this.claudeSessionId } : {}), ...this.config.extra?.claude, }; @@ -828,6 +819,19 @@ class ClaudeAgentSession implements AgentSession { sdkMessage: SDKUserMessage, turnId: number ): AsyncGenerator { + // If there's a pending interrupt, await it BEFORE calling ensureQuery(). + // interruptActiveTurn() clears this.query after interrupt() returns, + // so we must wait for it to complete before we try to get the query. + if (this.pendingInterruptPromise) { + await this.pendingInterruptPromise; + this.pendingInterruptPromise = null; + } + + // Check if we were superseded while waiting for the interrupt + if (this.currentTurnId !== turnId) { + return; + } + const query = await this.ensureQuery(); if (!this.input) { throw new Error("Claude session input stream not initialized"); @@ -837,7 +841,6 @@ class ClaudeAgentSession implements AgentSession { 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; } @@ -914,11 +917,19 @@ class ClaudeAgentSession implements AgentSession { } private async interruptActiveTurn(): Promise { - if (!this.query || typeof this.query.interrupt !== "function") { + const queryToInterrupt = this.query; + if (!queryToInterrupt || typeof queryToInterrupt.interrupt !== "function") { return; } try { - await this.query.interrupt(); + await queryToInterrupt.interrupt(); + // After interrupt(), the query iterator is done (returns done: true). + // Clear it so ensureQuery() creates a fresh query for the next turn. + // Also end the input stream and call return() to clean up the SDK process. + this.input?.end(); + await queryToInterrupt.return?.(); + this.query = null; + this.input = null; } catch (error) { console.warn("[ClaudeAgentSession] Failed to interrupt active turn:", error); } diff --git a/packages/server/src/server/agent/providers/claude-sdk-behavior.test.ts b/packages/server/src/server/agent/providers/claude-sdk-behavior.test.ts new file mode 100644 index 000000000..7d016b789 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude-sdk-behavior.test.ts @@ -0,0 +1,170 @@ +/** + * Direct SDK behavior tests - uses same setup as claude-agent.ts + */ +import { mkdtempSync, rmSync, realpathSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { describe, it, expect } from "vitest"; +import { + query, + type SDKMessage, + type SDKUserMessage, +} from "@anthropic-ai/claude-agent-sdk"; + +class Pushable implements AsyncIterable { + private queue: T[] = []; + private resolvers: Array<(value: IteratorResult) => void> = []; + private closed = false; + + push(item: T) { + if (this.closed) return; + if (this.resolvers.length > 0) { + this.resolvers.shift()!({ value: item, done: false }); + } else { + this.queue.push(item); + } + } + + end() { + this.closed = true; + while (this.resolvers.length > 0) { + this.resolvers.shift()!({ value: undefined, done: true }); + } + } + + [Symbol.asyncIterator](): AsyncIterator { + return { + next: (): Promise> => { + if (this.queue.length > 0) { + return Promise.resolve({ value: this.queue.shift()!, done: false }); + } + if (this.closed) { + return Promise.resolve({ value: undefined, done: true }); + } + return new Promise((resolve) => this.resolvers.push(resolve)); + }, + }; + } +} + +function tmpCwd(): string { + const dir = mkdtempSync(path.join(tmpdir(), "sdk-behavior-")); + try { + return realpathSync(dir); + } catch { + return dir; + } +} + +describe("Claude SDK direct behavior", () => { + it("shows what happens after interrupt()", async () => { + const cwd = tmpCwd(); + const input = new Pushable(); + + // Use same options as claude-agent.ts + const q = query({ + prompt: input, + options: { + cwd, + includePartialMessages: true, + permissionMode: "bypassPermissions", + systemPrompt: { + type: "preset", + preset: "claude_code", + }, + settingSources: ["user", "project"], + stderr: (data: string) => { + console.error("[SDK stderr]", data.trim()); + }, + }, + }); + + try { + // Send first message + console.log("[SDK] Sending MSG1"); + input.push({ + type: "user", + message: { role: "user", content: "Say exactly: MESSAGE_ONE" }, + parent_tool_use_id: null, + session_id: "", + }); + + // Collect events until we see assistant, then interrupt + const msg1Events: SDKMessage[] = []; + for await (const event of q) { + msg1Events.push(event); + console.log("[SDK] MSG1 event:", event.type); + + if (event.type === "assistant") { + // Push MSG2 BEFORE interrupt (like our wrapper does when a new message comes in) + console.log("[SDK] Pushing MSG2 before interrupt..."); + input.push({ + type: "user", + message: { role: "user", content: "Say exactly: MESSAGE_TWO" }, + parent_tool_use_id: null, + session_id: "", + }); + console.log("[SDK] Calling interrupt()..."); + await q.interrupt(); + console.log("[SDK] interrupt() returned"); + break; + } + if (event.type === "result") { + console.log("[SDK] MSG1 completed before interrupt"); + break; + } + } + + console.log("[SDK] MSG1 events:", msg1Events.length); + + // MSG2 was already pushed before interrupt + console.log("[SDK] About to call q.next() for MSG2..."); + const msg2Events: SDKMessage[] = []; + for await (const event of q) { + msg2Events.push(event); + + let detail = ""; + if (event.type === "assistant" && "message" in event && event.message?.content) { + const content = event.message.content; + if (Array.isArray(content)) { + for (const block of content) { + if (block.type === "text" && block.text) { + detail = `"${block.text.substring(0, 50)}"`; + } + } + } + } + console.log("[SDK] MSG2 event:", event.type, detail); + + if (event.type === "result") { + break; + } + } + console.log("[SDK] MSG2 loop finished"); + + // Analyze response + let responseText = ""; + for (const event of msg2Events) { + if (event.type === "assistant" && "message" in event && event.message?.content) { + const content = event.message.content; + if (Array.isArray(content)) { + for (const block of content) { + if (block.type === "text" && block.text) { + responseText += block.text; + } + } + } + } + } + + console.log("[SDK] MSG2 response:", responseText); + console.log("[SDK] Mentions 'one':", responseText.toLowerCase().includes("one")); + console.log("[SDK] Mentions 'two':", responseText.toLowerCase().includes("two")); + + expect(msg2Events.some(e => e.type === "result")).toBe(true); + } finally { + input.end(); + rmSync(cwd, { recursive: true, force: true }); + } + }, 120000); +}); diff --git a/packages/server/src/server/daemon.e2e.test.ts b/packages/server/src/server/daemon.e2e.test.ts index a53b0d1ff..95dc0ad29 100644 --- a/packages/server/src/server/daemon.e2e.test.ts +++ b/packages/server/src/server/daemon.e2e.test.ts @@ -23,7 +23,7 @@ describe("daemon E2E", () => { afterEach(async () => { await ctx.cleanup(); - }); + }, 60000); test("creates agent and receives response", async () => { // Create a Codex agent @@ -367,21 +367,22 @@ describe("daemon E2E", () => { expect(agent.id).toBeTruthy(); - // Ask Codex to sleep 60 seconds then write a file + // Ask Codex to sleep 15 seconds then write a file ctx.client.clearMessageQueue(); await ctx.client.sendMessage( agent.id, - "Run this bash command: sleep 60 && echo 'abort-test-completed' > abort-test-file.txt" + "Run this bash command: sleep 15 && echo 'abort-test-completed' > abort-test-file.txt" ); - // Wait 5 seconds for the command to start - await new Promise((r) => setTimeout(r, 5000)); + // Wait 3 seconds for the command to start + await new Promise((r) => setTimeout(r, 3000)); // Cancel/interrupt the agent await ctx.client.cancelAgent(agent.id); - // Wait 60 seconds (if abort works, the file should NOT be written) - await new Promise((r) => setTimeout(r, 60000)); + // Wait 10 seconds - if abort works, file should NOT be written + // (sleep would have completed at 15s if not interrupted) + await new Promise((r) => setTimeout(r, 10000)); // Assert the file was NOT created (proving Codex actually stopped) const fileExists = existsSync(filePath); @@ -389,7 +390,7 @@ describe("daemon E2E", () => { rmSync(cwd, { recursive: true, force: true }); }, - 90000 // 90 second timeout + 30000 // 30 second timeout ); // TODO: Fix this test - there's a race condition causing timeout errors @@ -3112,6 +3113,314 @@ describe("daemon E2E", () => { }, 180000 // 3 minute timeout ); + + test( + "sending message while agent is executing a tool call", + async () => { + const cwd = tmpCwd(); + + const agent = await ctx.client.createAgent({ + provider: "claude", + cwd, + title: "Interrupt During Tool Call Test", + modeId: "bypassPermissions", + }); + + expect(agent.id).toBeTruthy(); + + // Start a long-running tool call (sleep command) + console.log("[TOOL INTERRUPT TEST] Sending message 1 with long sleep command..."); + const msg1StartPosition = ctx.client.getMessageQueue().length; + + await ctx.client.sendMessage( + agent.id, + "Run: sleep 30 && echo 'done sleeping'" + ); + + // Wait for the tool call to start (agent should be running and we should see the bash tool) + console.log("[TOOL INTERRUPT TEST] Waiting for tool call to start..."); + let sawToolCall = false; + const toolWaitStart = Date.now(); + while (Date.now() - toolWaitStart < 30000) { + const queue = ctx.client.getMessageQueue(); + for (let i = msg1StartPosition; i < queue.length; i++) { + const m = queue[i]; + if ( + m.type === "agent_stream" && + m.payload.agentId === agent.id && + m.payload.event.type === "timeline" && + m.payload.event.item.type === "tool_call" + ) { + const tc = m.payload.event.item; + if (tc.name?.toLowerCase().includes("bash") || tc.name?.toLowerCase().includes("shell")) { + sawToolCall = true; + console.log("[TOOL INTERRUPT TEST] Tool call started:", tc.name); + break; + } + } + } + if (sawToolCall) break; + await new Promise(r => setTimeout(r, 200)); + } + + expect(sawToolCall).toBe(true); + + // Now send an interrupting message while the tool is running + console.log("[TOOL INTERRUPT TEST] Sending message 2 to interrupt..."); + const msg2StartPosition = ctx.client.getMessageQueue().length; + + await ctx.client.sendMessage( + agent.id, + "Stop. Say exactly: 'interrupted tool call'" + ); + + // Track state transitions + console.log("[TOOL INTERRUPT TEST] Monitoring state transitions..."); + const stateTransitions: Array<{ status: string; timestamp: number; lastError?: string }> = []; + const monitorStart = Date.now(); + + while (Date.now() - monitorStart < 60000) { + const queue = ctx.client.getMessageQueue(); + for (let i = msg2StartPosition; i < queue.length; i++) { + const m = queue[i]; + if (m.type === "agent_state" && m.payload.id === agent.id) { + const lastRecorded = stateTransitions[stateTransitions.length - 1]; + if (!lastRecorded || lastRecorded.status !== m.payload.status) { + stateTransitions.push({ + status: m.payload.status, + timestamp: Date.now() - monitorStart, + lastError: m.payload.lastError, + }); + console.log(`[TOOL INTERRUPT TEST] State: ${m.payload.status} at ${Date.now() - monitorStart}ms`, m.payload.lastError || ""); + } + } + } + + // Check if we reached idle/error after seeing running + const sawRunning = stateTransitions.some(s => s.status === "running"); + const sawFinal = stateTransitions.some(s => s.status === "idle" || s.status === "error"); + if (sawRunning && sawFinal) { + break; + } + + await new Promise(r => setTimeout(r, 200)); + } + + console.log("[TOOL INTERRUPT TEST] State transitions:", JSON.stringify(stateTransitions, null, 2)); + + // Verify we got proper state transitions + expect(stateTransitions.length).toBeGreaterThan(0); + + // Check if we ended in idle (success) or error + const finalState = stateTransitions[stateTransitions.length - 1]; + console.log("[TOOL INTERRUPT TEST] Final state:", finalState); + + // Look for the response to message 2 + const queue = ctx.client.getMessageQueue(); + const assistantMessages: string[] = []; + let foundMsg2User = false; + + for (let i = msg2StartPosition; i < queue.length; i++) { + const m = queue[i]; + 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("interrupted tool call")) { + foundMsg2User = true; + } + if (foundMsg2User && item.type === "assistant_message" && item.text) { + assistantMessages.push(item.text); + } + } + } + + console.log("[TOOL INTERRUPT TEST] Found user message 2:", foundMsg2User); + console.log("[TOOL INTERRUPT TEST] Assistant messages after msg2:", assistantMessages); + + // The key assertion: agent should have responded to message 2 + if (finalState.status === "idle") { + expect(assistantMessages.length).toBeGreaterThan(0); + } + + await ctx.client.deleteAgent(agent.id); + rmSync(cwd, { recursive: true, force: true }); + }, + 180000 + ); + + test( + "rapid sequential messages to same agent", + async () => { + const cwd = tmpCwd(); + + const agent = await ctx.client.createAgent({ + provider: "claude", + cwd, + title: "Rapid Sequential Messages Test", + modeId: "bypassPermissions", + }); + + expect(agent.id).toBeTruthy(); + + // Send 3 messages in rapid succession without waiting + console.log("[RAPID MSG TEST] Sending 3 messages rapidly..."); + const startPosition = ctx.client.getMessageQueue().length; + + const msg1 = "Say: MESSAGE_ONE"; + const msg2 = "Say: MESSAGE_TWO"; + const msg3 = "Say: MESSAGE_THREE"; + + // Helper to count user messages in queue + const countUserMessages = (): number => { + let count = 0; + const queue = ctx.client.getMessageQueue(); + for (let i = startPosition; i < queue.length; i++) { + const m = queue[i]; + if ( + m.type === "agent_stream" && + m.payload.agentId === agent.id && + m.payload.event.type === "timeline" && + m.payload.event.item.type === "user_message" + ) { + count++; + } + } + return count; + }; + + // Send all 3 messages + await ctx.client.sendMessage(agent.id, msg1); + console.log("[RAPID MSG TEST] MSG1 sent, waiting briefly..."); + await new Promise(r => setTimeout(r, 100)); + + await ctx.client.sendMessage(agent.id, msg2); + console.log("[RAPID MSG TEST] MSG2 sent, waiting briefly..."); + await new Promise(r => setTimeout(r, 100)); + + await ctx.client.sendMessage(agent.id, msg3); + console.log("[RAPID MSG TEST] MSG3 sent"); + + // First, wait until all 3 user messages are recorded + console.log("[RAPID MSG TEST] Waiting for all 3 user messages to be recorded..."); + const userMsgWaitStart = Date.now(); + while (Date.now() - userMsgWaitStart < 30000) { + const count = countUserMessages(); + console.log(`[RAPID MSG TEST] User message count: ${count}`); + if (count >= 3) break; + await new Promise(r => setTimeout(r, 200)); + } + + const finalUserMsgCount = countUserMessages(); + console.log(`[RAPID MSG TEST] Final user message count: ${finalUserMsgCount}`); + + // Now wait for agent to become idle after processing all 3 messages + // We need to see at least 3 running transitions to know all messages were processed + console.log("[RAPID MSG TEST] Waiting for agent to finish processing all messages..."); + const waitStart = Date.now(); + let runningCount = 0; + let finalState: AgentSnapshotPayload | null = null; + + while (Date.now() - waitStart < 120000) { + const queue = ctx.client.getMessageQueue(); + let currentRunningCount = 0; + let lastState: AgentSnapshotPayload | null = null; + + for (let i = startPosition; i < queue.length; i++) { + const m = queue[i]; + if (m.type === "agent_state" && m.payload.id === agent.id) { + if (m.payload.status === "running") currentRunningCount++; + lastState = m.payload; + } + } + + runningCount = currentRunningCount; + + // Need to have seen at least 3 running states (one per message) and end up idle + if (runningCount >= 3 && lastState && lastState.status === "idle") { + finalState = lastState; + break; + } + await new Promise(r => setTimeout(r, 500)); + } + + console.log("[RAPID MSG TEST] Final state:", finalState?.status, finalState?.lastError); + console.log("[RAPID MSG TEST] Total running transitions:", runningCount); + + // Analyze what happened + const queue = ctx.client.getMessageQueue(); + const userMessages: string[] = []; + const assistantMessages: string[] = []; + const stateChanges: string[] = []; + + for (let i = startPosition; i < queue.length; i++) { + const m = queue[i]; + if (m.type === "agent_state" && m.payload.id === agent.id) { + stateChanges.push(m.payload.status); + } + 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") { + userMessages.push((item.text as string) || ""); + } + if (item.type === "assistant_message" && item.text) { + assistantMessages.push(item.text); + } + } + } + + console.log("[RAPID MSG TEST] User messages recorded:", userMessages.length, userMessages); + console.log("[RAPID MSG TEST] Assistant messages:", assistantMessages.length, JSON.stringify(assistantMessages)); + console.log("[RAPID MSG TEST] State changes:", stateChanges.join(" -> ")); + console.log("[RAPID MSG TEST] Total queue items since start:", queue.length - startPosition); + + // All 3 user messages should have been recorded + expect(userMessages.length).toBe(3); + + // Agent should have responded (at least to the final message) + expect(assistantMessages.length).toBeGreaterThan(0); + + // The last turn should have completed successfully (not failed due to race condition) + const lastResponse = assistantMessages[assistantMessages.length - 1]?.toLowerCase() || ""; + console.log("[RAPID MSG TEST] Last response:", lastResponse); + + // Verify we got a proper turn_completed event (not turn_failed from race condition) + const turnCompletedEvents = queue.filter((m, i) => + i >= startPosition && + m.type === "agent_stream" && + m.payload.agentId === agent.id && + m.payload.event.type === "turn_completed" + ); + const turnFailedEvents = queue.filter((m, i) => + i >= startPosition && + m.type === "agent_stream" && + m.payload.agentId === agent.id && + m.payload.event.type === "turn_failed" + ); + console.log("[RAPID MSG TEST] Turn completed events:", turnCompletedEvents.length); + console.log("[RAPID MSG TEST] Turn failed events:", turnFailedEvents.length); + + // The final turn should complete successfully, not fail + expect(turnCompletedEvents.length).toBeGreaterThan(0); + // We might have some turn_failed from interrupted turns, but the last turn should succeed + expect(turnFailedEvents.length).toBe(0); + + // The response should mention "three" since that was the last message sent + const combinedResponse = assistantMessages.join(" ").toLowerCase(); + console.log("[RAPID MSG TEST] Combined response:", combinedResponse); + expect(combinedResponse).toContain("three"); + + await ctx.client.deleteAgent(agent.id); + rmSync(cwd, { recursive: true, force: true }); + }, + 180000 + ); }); describe("tool call structure", () => {