fix(orchestrate): do not wait on a dead SSE stream after the watchdog has the result

orchestrate/skills/orchestrate/scripts/__tests__/wait-handoff-dead-stream.test.ts

@@ -0,0 +1,107 @@
+import { afterAll, beforeAll, describe, expect, mock, test } from "bun:test";
+import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+// A run whose SSE has died: `stream()` never yields again and never ends, and
+// `wait()` (which rides the same stream) never settles. The REST side has
+// already moved the run to `finished`, which is what the watchdog poll sees.
+
+const SDK_MOCK = "@cursor/sdk";
+
+const finishedResult = {
+  id: "run-dead-sse",
+  status: "finished" as const,
+  result: "## Status\nhanded-off\n\n## Branch\nagent/worker-one\n",
+  durationMs: 10_000,
+  git: { branches: [{ branch: "agent/worker-one" }] },
+};
+
+const never = () => new Promise<never>(() => {});
+
+function deadStreamRun(status: "running" | "finished") {
+  return {
+    id: "run-dead-sse",
+    agentId: "bc-dead-sse",
+    status,
+    stream: async function* () {
+      yield {
+        type: "assistant",
+        message: { content: [{ type: "text", text: "working..." }] },
+      };
+      await never();
+    },
+    wait: status === "finished" ? async () => finishedResult : never,
+  };
+}
+
+mock.module(SDK_MOCK, () => ({
+  Agent: {
+    create: async () => ({
+      agentId: "bc-dead-sse",
+      send: async () => deadStreamRun("running"),
+      [Symbol.asyncDispose]: () => Promise.resolve(),
+    }),
+    // The poll sees the server-side truth.
+    getRun: async () => deadStreamRun("finished"),
+  },
+  CursorAgentError: class CursorAgentError extends Error {},
+}));
+
+const { AgentManager, __resetSDKForTests } = await import(
+  "../core/agent-manager.ts"
+);
+__resetSDKForTests();
+
+const ORIGINAL_API_KEY = process.env.CURSOR_API_KEY;
+const ORIGINAL_POLL = process.env.ORCHESTRATE_WATCHDOG_POLL_MS;
+
+beforeAll(() => {
+  process.env.CURSOR_API_KEY = "test-key";
+  process.env.ORCHESTRATE_WATCHDOG_POLL_MS = "50";
+});
+
+afterAll(() => {
+  if (ORIGINAL_API_KEY === undefined) delete process.env.CURSOR_API_KEY;
+  else process.env.CURSOR_API_KEY = ORIGINAL_API_KEY;
+  if (ORIGINAL_POLL === undefined)
+    delete process.env.ORCHESTRATE_WATCHDOG_POLL_MS;
+  else process.env.ORCHESTRATE_WATCHDOG_POLL_MS = ORIGINAL_POLL;
+});
+
+describe("waitAndHandoff with a dead SSE stream", () => {
+  test("Hands the task off from the watchdog's result instead of waiting on the stream forever", async () => {
+    const workspace = mkdtempSync(join(tmpdir(), "orch-dead-sse-"));
+    writeFileSync(
+      join(workspace, "plan.json"),
+      JSON.stringify({
+        goal: "survive a dead stream",
+        rootSlug: "dead-sse",
+        baseBranch: "main",
+        repoUrl: "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/example-org/example-repo",
+        tasks: [
+          { name: "worker-one", type: "worker", scopedGoal: "Do the work." },
+        ],
+      })
+    );
+    try {
+      const mgr = await AgentManager.load(workspace);
+      const def = mgr.plan.tasks?.find(t => t.name === "worker-one");
+      if (!def) throw new Error("missing task def");
+      const spawned = await mgr.spawnTask(def);
+      if (!spawned) throw new Error("spawn failed");
+
+      const outcome = await Promise.race([
+        mgr.waitAndHandoff(spawned).then(() => "handed-off" as const),
+        new Promise<"hung">(resolve =>
+          setTimeout(() => resolve("hung"), 8_000)
+        ),
+      ]);
+
+      expect(outcome).toBe("handed-off");
+      expect(mgr.getTask("worker-one")?.status).toBe("handed-off");
+    } finally {
+      rmSync(workspace, { recursive: true, force: true });
+    }
+  }, 15_000);
+});

orchestrate/skills/orchestrate/scripts/core/agent-manager.ts

@@ -41,6 +41,7 @@ import type { CommentDestinations } from "./comment-retry-queue.ts";
 
 const SPAWN_MAX_ATTEMPTS = 3;
 const SPAWN_RETRY_BACKOFF_MS = 2_000;
+// Overridable for tests via ORCHESTRATE_WATCHDOG_POLL_MS.
 const WAIT_WATCHDOG_POLL_INTERVAL_MS = 60_000;
 // Internal watchdog "noticed idle" debug logs. Lower threshold = earlier
 // visibility into stalled SSE / tool-call streams without disturbing the
@@ -55,6 +56,10 @@ const STUCK_ESCALATION_MS = 1_800_000;
 // still reports running. Without it, state.attention and saveState frequency
 // would blow up. Negligible cost when retries take ~1-2 minutes.
 const WAIT_RECOVERY_RETRY_BACKOFF_MS = 5_000;
+// How long waitAndHandoff waits for `run.stream()` to end after `run.wait()`
+// (or the watchdog poll) has already produced the result. A live stream ends
+// within milliseconds of the run; a dead one never does.
+const WAIT_STREAM_DRAIN_GRACE_MS = 2_000;
 
 export type RespawnSource = "local-cli" | "self-planner" | "script-auto-retry";
 
@@ -883,7 +888,12 @@ export class AgentManager {
           lastWaitError = errorMessage(waitErr);
           throw waitErr;
         }
-        await streamPromise;
+        // When the watchdog poll won the race, `run.stream()` is still parked
+        // on a dead SSE and never ends; awaiting it here would hang the
+        // reconcile loop that the watchdog exists to free. `rr` is
+        // authoritative either way, so the stream gets a short grace to
+        // deliver what it already has, then the loop moves on without it.
+        await Promise.race([streamPromise, sleep(WAIT_STREAM_DRAIN_GRACE_MS)]);
         if (rr.status !== "error") break;
         lastWaitError = runResultErrorMessage(rr) ?? lastWaitError;
 
@@ -1864,7 +1874,7 @@ async function waitRunWithWatchdog(args: {
   const pollLoop = async (): Promise<RunResult> => {
     while (!ac.signal.aborted) {
       try {
-        await sleep(WAIT_WATCHDOG_POLL_INTERVAL_MS, undefined, {
+        await sleep(watchdogPollIntervalMs(), undefined, {
           signal: ac.signal,
         });
       } catch {
@@ -2013,6 +2023,15 @@ export function formatToolCallIdleWarning(args: {
   return `${args.taskLabel}: tool_call idle ${args.idleMs}ms; last=${lastCall}`;
 }
 
+function watchdogPollIntervalMs(): number {
+  const raw = process.env.ORCHESTRATE_WATCHDOG_POLL_MS;
+  if (!raw) return WAIT_WATCHDOG_POLL_INTERVAL_MS;
+  const parsed = Number.parseInt(raw, 10);
+  return Number.isFinite(parsed) && parsed > 0
+    ? parsed
+    : WAIT_WATCHDOG_POLL_INTERVAL_MS;
+}
+
 function toolCallStuckThresholdMs(): number {
   const raw = process.env.ORCHESTRATE_TOOL_CALL_IDLE_MS;
   if (!raw) return STUCK_ESCALATION_MS;