diff --git a/orchestrate/skills/orchestrate/scripts/__tests__/wait-handoff-dead-stream.test.ts b/orchestrate/skills/orchestrate/scripts/__tests__/wait-handoff-dead-stream.test.ts new file mode 100644 index 000000000..9bad5b1ca --- /dev/null +++ b/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(() => {}); + +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); +}); diff --git a/orchestrate/skills/orchestrate/scripts/core/agent-manager.ts b/orchestrate/skills/orchestrate/scripts/core/agent-manager.ts index 32ce7199f..0c7808196 100644 --- a/orchestrate/skills/orchestrate/scripts/core/agent-manager.ts +++ b/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 => { 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;