From efde557d6f1de0273909cdb64013500a22966dfd Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Tue, 17 Feb 2026 10:43:20 +0700 Subject: [PATCH] Fix stuck send-while-running processing recovery --- .../app/src/components/agent-input-area.tsx | 27 ++++- ...hile-running-stuck-claude.real.e2e.test.ts | 107 ++++++++++++++++++ .../send-while-running-stuck-test-utils.ts | 42 +++++++ .../send-while-running-stuck.real.e2e.test.ts | 104 +++++++++++++++++ packages/server/src/server/session.ts | 26 +++++ 5 files changed, 304 insertions(+), 2 deletions(-) create mode 100644 packages/server/src/server/daemon-e2e/send-while-running-stuck-claude.real.e2e.test.ts create mode 100644 packages/server/src/server/daemon-e2e/send-while-running-stuck-test-utils.ts create mode 100644 packages/server/src/server/daemon-e2e/send-while-running-stuck.real.e2e.test.ts diff --git a/packages/app/src/components/agent-input-area.tsx b/packages/app/src/components/agent-input-area.tsx index 622f2b48d..2f35633e3 100644 --- a/packages/app/src/components/agent-input-area.tsx +++ b/packages/app/src/components/agent-input-area.tsx @@ -233,15 +233,38 @@ export function AgentInputArea({ }, [onSubmitMessage]); const isAgentRunning = agent?.status === "running"; + const agentUpdatedAtMs = agent?.updatedAt?.getTime() ?? 0; const prevIsAgentRunningRef = useRef(isAgentRunning); + const latestAgentUpdatedAtRef = useRef(agentUpdatedAtMs); useEffect(() => { + const previousUpdatedAt = latestAgentUpdatedAtRef.current; + if (agentUpdatedAtMs < previousUpdatedAt) { + return; + } + const wasRunning = prevIsAgentRunningRef.current; + let shouldClearProcessing = false; + + if (isProcessing) { + const hasEnteredRunning = !wasRunning && isAgentRunning; + const hasFreshRunningUpdateWhileRunning = + wasRunning && isAgentRunning && agentUpdatedAtMs > previousUpdatedAt; + const hasStoppedRunning = wasRunning && !isAgentRunning; + + shouldClearProcessing = + hasEnteredRunning || + hasFreshRunningUpdateWhileRunning || + hasStoppedRunning; + } + prevIsAgentRunningRef.current = isAgentRunning; - if (!wasRunning && isAgentRunning && isProcessing) { + latestAgentUpdatedAtRef.current = agentUpdatedAtMs; + + if (shouldClearProcessing) { setIsProcessing(false); } - }, [isAgentRunning, isProcessing]); + }, [agentUpdatedAtMs, isAgentRunning, isProcessing]); const updateQueue = useCallback( (updater: (current: QueuedMessage[]) => QueuedMessage[]) => { diff --git a/packages/server/src/server/daemon-e2e/send-while-running-stuck-claude.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/send-while-running-stuck-claude.real.e2e.test.ts new file mode 100644 index 000000000..98ae27b53 --- /dev/null +++ b/packages/server/src/server/daemon-e2e/send-while-running-stuck-claude.real.e2e.test.ts @@ -0,0 +1,107 @@ +import { describe, test, expect } from "vitest"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import pino from "pino"; + +import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js"; +import { DaemonClient } from "../test-utils/daemon-client.js"; +import { ClaudeAgentClient } from "../agent/providers/claude-agent.js"; +import { getFullAccessConfig } from "./agent-configs.js"; +import { applyAgentInputProcessingTransition } from "./send-while-running-stuck-test-utils.js"; +import { isCommandAvailable } from "../agent/provider-launch-config.js"; + +function tmpCwd(): string { + return mkdtempSync(path.join(tmpdir(), "daemon-real-stuck-claude-")); +} + +describe("daemon E2E (real claude) - send while running recovery", () => { + test.runIf(isCommandAvailable("claude"))( + "clears input processing when the interrupt transition is missed", + async () => { + const logger = pino({ level: "silent" }); + const cwd = tmpCwd(); + const daemon = await createTestPaseoDaemon({ + agentClients: { claude: new ClaudeAgentClient({ logger }) }, + logger, + }); + + const primary = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + const secondary = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + + try { + await primary.connect(); + await secondary.connect(); + primary.subscribeAgentUpdates({ subscriptionId: "primary" }); + secondary.subscribeAgentUpdates({ subscriptionId: "secondary" }); + + const agent = await primary.createAgent({ + cwd, + title: "stuck-repro-real-claude", + ...getFullAccessConfig("claude"), + }); + + await primary.sendMessage( + agent.id, + "Run bash command sleep 30, wait for completion, then reply done." + ); + await primary.waitForAgentUpsert( + agent.id, + (snapshot) => snapshot.status === "running", + 60_000 + ); + + let isProcessing = true; + let previousIsRunning = true; + let latestUpdatedAt = Date.now(); + + await primary.close(); + + await secondary.sendMessage(agent.id, "Reply with exactly: state saved"); + await secondary.waitForAgentUpsert( + agent.id, + (snapshot) => snapshot.status === "running", + 60_000 + ); + + const reconnected = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + try { + await reconnected.connect(); + reconnected.subscribeAgentUpdates({ subscriptionId: "reconnected" }); + + reconnected.on("agent_update", (message) => { + if (message.type !== "agent_update" || message.payload.kind !== "upsert") { + return; + } + if (message.payload.agent.id !== agent.id) { + return; + } + + const next = applyAgentInputProcessingTransition({ + snapshot: message.payload.agent, + currentIsProcessing: isProcessing, + previousIsRunning, + latestUpdatedAt, + }); + isProcessing = next.isProcessing; + previousIsRunning = next.previousIsRunning; + latestUpdatedAt = next.latestUpdatedAt; + }); + + await secondary.waitForFinish(agent.id, 180_000); + + // Sending while running should clear processing even if reconnect misses the + // not-running -> running transition. + expect(isProcessing).toBe(false); + } finally { + await reconnected.close(); + } + } finally { + await secondary.close(); + await daemon.close(); + rmSync(cwd, { recursive: true, force: true }); + } + }, + 300_000 + ); +}); diff --git a/packages/server/src/server/daemon-e2e/send-while-running-stuck-test-utils.ts b/packages/server/src/server/daemon-e2e/send-while-running-stuck-test-utils.ts new file mode 100644 index 000000000..0e148c6a4 --- /dev/null +++ b/packages/server/src/server/daemon-e2e/send-while-running-stuck-test-utils.ts @@ -0,0 +1,42 @@ +import type { AgentSnapshotPayload } from "../messages.js"; + +export function applyAgentInputProcessingTransition(input: { + snapshot: AgentSnapshotPayload; + currentIsProcessing: boolean; + previousIsRunning: boolean; + latestUpdatedAt: number; +}): { isProcessing: boolean; previousIsRunning: boolean; latestUpdatedAt: number } { + const updatedAt = new Date(input.snapshot.updatedAt).getTime(); + if (updatedAt < input.latestUpdatedAt) { + return { + isProcessing: input.currentIsProcessing, + previousIsRunning: input.previousIsRunning, + latestUpdatedAt: input.latestUpdatedAt, + }; + } + + const isRunning = input.snapshot.status === "running"; + const wasRunning = input.previousIsRunning; + let isProcessing = input.currentIsProcessing; + + if (isProcessing) { + const hasEnteredRunning = !wasRunning && isRunning; + const hasFreshRunningUpdateWhileRunning = + wasRunning && isRunning && updatedAt > input.latestUpdatedAt; + const hasStoppedRunning = wasRunning && !isRunning; + + if ( + hasEnteredRunning || + hasFreshRunningUpdateWhileRunning || + hasStoppedRunning + ) { + isProcessing = false; + } + } + + return { + isProcessing, + previousIsRunning: isRunning, + latestUpdatedAt: updatedAt, + }; +} diff --git a/packages/server/src/server/daemon-e2e/send-while-running-stuck.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/send-while-running-stuck.real.e2e.test.ts new file mode 100644 index 000000000..d2a21c728 --- /dev/null +++ b/packages/server/src/server/daemon-e2e/send-while-running-stuck.real.e2e.test.ts @@ -0,0 +1,104 @@ +import { describe, test, expect } from "vitest"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import pino from "pino"; + +import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js"; +import { DaemonClient } from "../test-utils/daemon-client.js"; +import { CodexAppServerAgentClient } from "../agent/providers/codex-app-server-agent.js"; +import { getFullAccessConfig } from "./agent-configs.js"; +import { applyAgentInputProcessingTransition } from "./send-while-running-stuck-test-utils.js"; +import { isCommandAvailable } from "../agent/provider-launch-config.js"; + +function tmpCwd(): string { + return mkdtempSync(path.join(tmpdir(), "daemon-real-stuck-")); +} + +describe("daemon E2E (real codex) - send while running recovery", () => { + test.runIf(isCommandAvailable("codex"))( + "clears input processing when the interrupt transition is missed", + async () => { + const logger = pino({ level: "silent" }); + const cwd = tmpCwd(); + const daemon = await createTestPaseoDaemon({ + agentClients: { codex: new CodexAppServerAgentClient(logger) }, + logger, + }); + + const primary = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + const secondary = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + + try { + await primary.connect(); + await secondary.connect(); + primary.subscribeAgentUpdates({ subscriptionId: "primary" }); + secondary.subscribeAgentUpdates({ subscriptionId: "secondary" }); + + const agent = await primary.createAgent({ + cwd, + title: "stuck-repro-real-codex", + ...getFullAccessConfig("codex"), + }); + + await primary.sendMessage(agent.id, "Run: sleep 30"); + await primary.waitForAgentUpsert( + agent.id, + (snapshot) => snapshot.status === "running", + 30_000 + ); + + let isProcessing = true; + let previousIsRunning = true; + let latestUpdatedAt = Date.now(); + + await primary.close(); + + await secondary.sendMessage(agent.id, "Reply with exactly: state saved"); + await secondary.waitForAgentUpsert( + agent.id, + (snapshot) => snapshot.status === "running", + 30_000 + ); + + const reconnected = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + try { + await reconnected.connect(); + reconnected.subscribeAgentUpdates({ subscriptionId: "reconnected" }); + + reconnected.on("agent_update", (message) => { + if (message.type !== "agent_update" || message.payload.kind !== "upsert") { + return; + } + if (message.payload.agent.id !== agent.id) { + return; + } + + const next = applyAgentInputProcessingTransition({ + snapshot: message.payload.agent, + currentIsProcessing: isProcessing, + previousIsRunning, + latestUpdatedAt, + }); + isProcessing = next.isProcessing; + previousIsRunning = next.previousIsRunning; + latestUpdatedAt = next.latestUpdatedAt; + }); + + await secondary.waitForFinish(agent.id, 120_000); + + // Sending while running should clear processing even if reconnect misses the + // not-running -> running transition. + expect(isProcessing).toBe(false); + } finally { + await reconnected.close(); + } + } finally { + await secondary.close(); + await daemon.close(); + rmSync(cwd, { recursive: true, force: true }); + } + }, + 240_000 + ); +}); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index f0f73c338..3ba3ded09 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -1183,6 +1183,31 @@ export class Session { } } + private async emitCurrentAgentUpdatesForSubscription(): Promise { + const subscription = this.agentUpdatesSubscription; + if (!subscription) { + return; + } + + try { + const agents = await this.listAgentPayloads({ + labels: subscription.filter?.labels, + }); + for (const agent of agents) { + const project = await this.buildProjectPlacement(agent.cwd); + this.emit({ + type: "agent_update", + payload: { kind: "upsert", agent, project }, + }); + } + } catch (error) { + this.sessionLogger.error( + { err: error }, + "Failed to emit current agent updates for subscription bootstrap" + ); + } + } + /** * Main entry point for processing session messages */ @@ -1214,6 +1239,7 @@ export class Session { subscriptionId: msg.subscriptionId, filter: msg.filter, }; + await this.emitCurrentAgentUpdatesForSubscription(); break; case "unsubscribe_agent_updates":