Fix stuck send-while-running processing recovery

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
+  );
+});

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,
+  };
+}

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
+  );
+});

packages/server/src/server/session.ts

@@ -1183,6 +1183,31 @@ export class Session {
     }
   }
 
+  private async emitCurrentAgentUpdatesForSubscription(): Promise<void> {
+    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":