From 7895f2883d189217560d99870c554ac705d95f3f Mon Sep 17 00:00:00 2001 From: Emil Yahiaiev <154777503+emilaiev@users.noreply.github.com> Date: Sat, 24 Oct 2026 23:40:24 +0700 Subject: [PATCH] fix(provider-cursor): find the abandoned run when resuming a restarted thread If T3 stops while a Cursor run is streaming, the run stays non-terminal in the SDK's local store and every later send to that agent fails with "already has active run". The recovery for exactly this case scans the run list with limit: 1, but the SDK lists runs oldest first, so page 1 holds the first turn, never the still-running run. The scan finds nothing, rethrows the original error, and the thread stays stuck until someone cancels the orphan by hand. Page through the run list until the running run is found, and log the scan either way so a failed recovery is visible in the provider log. The test mock now mirrors the SDK: runs come back oldest first and limit truncates the page. A second test forces one-item pages so the cursor path is covered. --- .../src/server/CursorAgentSdk.test.ts | 73 +++++++++++++++++-- .../src/server/CursorAgentSdk.ts | 65 +++++++++++++++-- 2 files changed, 124 insertions(+), 14 deletions(-) diff --git a/packages/provider-cursor/src/server/CursorAgentSdk.test.ts b/packages/provider-cursor/src/server/CursorAgentSdk.test.ts index d715766a309bc..e9168f50f9457 100644 --- a/packages/provider-cursor/src/server/CursorAgentSdk.test.ts +++ b/packages/provider-cursor/src/server/CursorAgentSdk.test.ts @@ -14,6 +14,7 @@ const cursorSdkMock = vi.hoisted(() => { const runOperations: Array = []; let agentCloseFailure: unknown; let staleRunActive = false; + let listRunsPageSize: number | undefined; const runWait = vi.fn(async () => ({ id: "run-cursor-agent-sdk-test", requestId: "request-cursor-agent-sdk-test", @@ -62,12 +63,35 @@ const cursorSdkMock = vi.hoisted(() => { send, close: agentClose, })); - const listRuns = vi.fn(async () => { - runOperations.push("agent.listRuns"); - return { - items: staleRunActive ? [{ id: "run-stale", status: "running" }] : [], - }; - }); + // Mirrors the real SDK: runs come back oldest first, and a limit truncates the + // page instead of selecting the newest run. + const listRuns = vi.fn( + async ( + _agentId: string, + options?: { + readonly limit?: number; + readonly cursor?: string; + readonly runtime?: "local"; + readonly cwd?: string; + }, + ) => { + runOperations.push("agent.listRuns"); + const items = staleRunActive + ? [ + { id: "run-cursor-agent-sdk-test-earlier", status: "finished" }, + { id: "run-stale", status: "running" }, + ] + : []; + const offset = options?.cursor === undefined ? 0 : Number(options.cursor); + const pageSize = listRunsPageSize ?? options?.limit ?? items.length; + const page = items.slice(offset, offset + pageSize); + const nextOffset = offset + page.length; + return { + items: page, + ...(nextOffset < items.length ? { nextCursor: String(nextOffset) } : {}), + }; + }, + ); const cancelRun = vi.fn(async () => { runOperations.push("agent.cancelRun"); staleRunActive = false; @@ -87,6 +111,9 @@ const cursorSdkMock = vi.hoisted(() => { setAgentCloseFailure: (failure: unknown) => { agentCloseFailure = failure; }, + setListRunsPageSize: (size: number | undefined) => { + listRunsPageSize = size; + }, setStaleRunActive: (active: boolean) => { staleRunActive = active; }, @@ -161,7 +188,7 @@ describe("CursorAgentSdkRunner", () => { ]); assert.deepStrictEqual(cursorSdkMock.listRuns.mock.calls[0], [ "agent-cursor-agent-sdk-test", - { runtime: "local", cwd: process.cwd(), limit: 1 }, + { runtime: "local", cwd: process.cwd() }, ]); assert.deepStrictEqual(cursorSdkMock.cancelRun.mock.calls[0], [ "run-stale", @@ -172,4 +199,36 @@ describe("CursorAgentSdkRunner", () => { ]); }).pipe(Effect.provide(layerTest)), ); + + it.effect("finds the abandoned run when the store pages the run list", () => + Effect.gen(function* () { + cursorSdkMock.runOperations.length = 0; + cursorSdkMock.listRuns.mockClear(); + cursorSdkMock.cancelRun.mockClear(); + cursorSdkMock.setStaleRunActive(true); + cursorSdkMock.setListRunsPageSize(1); + + const runner = yield* CursorAgentSdk.CursorAgentSdkRunner; + const session = yield* runner.open({ + operation: "resume", + agentId: "agent-cursor-agent-sdk-test", + options: { model: { id: "default" }, mode: "agent", local: { cwd: process.cwd() } }, + threadId: ThreadId.make("thread-cursor-agent-sdk-stale-run-paged-test"), + providerSessionId: ProviderSessionId.make( + "provider-session-cursor-agent-sdk-stale-run-paged-test", + ), + }); + const run = yield* session.send({ message: "continue" }); + yield* run.wait; + + assert.deepStrictEqual(cursorSdkMock.runOperations, [ + "run.start", + "agent.listRuns", + "agent.listRuns", + "agent.cancelRun", + "run.start", + ]); + cursorSdkMock.setListRunsPageSize(undefined); + }).pipe(Effect.provide(layerTest)), + ); }); diff --git a/packages/provider-cursor/src/server/CursorAgentSdk.ts b/packages/provider-cursor/src/server/CursorAgentSdk.ts index 48c85af25487f..bf6bcc8e96b59 100644 --- a/packages/provider-cursor/src/server/CursorAgentSdk.ts +++ b/packages/provider-cursor/src/server/CursorAgentSdk.ts @@ -196,6 +196,16 @@ export type CursorAgentSdkProtocolLogEvent = readonly runId: string; }; } + | { + readonly direction: "incoming"; + readonly stage: "decoded"; + readonly payload: { + readonly type: "agent.runs.scan"; + readonly agentId: string; + readonly runCount: number; + readonly activeRunId?: string; + }; + } | { readonly direction: "outgoing"; readonly stage: "decoded"; @@ -236,6 +246,9 @@ function isActiveRunConflict(cause: unknown): boolean { return cause instanceof Error && /already has active run/i.test(cause.message); } +/** Bounds the abandoned-run scan; the SDK returns one page unless a cursor says otherwise. */ +const ABANDONED_RUN_MAX_SCAN_PAGES = 50; + export function isCursorCancellationError(cause: unknown): boolean { let current = cause; const seen = new Set(); @@ -466,19 +479,57 @@ export const makeCursorAgentSdkRunner = Effect.fn("makeCursorAgentSdkRunner")(fu }), catch: (cause) => runnerError(cause, "run.start"), }); + // The SDK lists runs oldest first, so asking for the first page finds the + // first turn rather than the run still marked active. The abandoned run + // that blocks a resumed send can sit anywhere in the list: page through + // until it is found, and record what the scan saw either way. + const findAbandonedRun = Effect.gen(function* () { + let cursor: string | undefined; + let runCount = 0; + for (let page = 0; page < ABANDONED_RUN_MAX_SCAN_PAGES; page += 1) { + const listing = yield* Effect.tryPromise({ + try: () => + Agent.listRuns(agent.agentId, { + ...runOptions, + ...(cursor === undefined ? {} : { cursor }), + }), + catch: (cause) => runnerError(cause, "agent.listRuns"), + }); + runCount += listing.items.length; + const activeRun = listing.items.find((candidate) => candidate.status === "running"); + if (activeRun !== undefined) { + yield* log({ + direction: "incoming", + stage: "decoded", + payload: { + type: "agent.runs.scan", + agentId: agent.agentId, + runCount, + activeRunId: activeRun.id, + }, + }); + return activeRun; + } + if (listing.nextCursor === undefined) { + break; + } + cursor = listing.nextCursor; + } + yield* log({ + direction: "incoming", + stage: "decoded", + payload: { type: "agent.runs.scan", agentId: agent.agentId, runCount }, + }); + return undefined; + }); + const run = yield* startRun().pipe( Effect.catchIf( (error) => abandonedRunRecoveryAvailable && isActiveRunConflict(error.cause), (error) => Effect.gen(function* () { abandonedRunRecoveryAvailable = false; - const latestRun = yield* Effect.tryPromise({ - try: () => Agent.listRuns(agent.agentId, { ...runOptions, limit: 1 }), - catch: (cause) => runnerError(cause, "agent.listRuns"), - }); - const activeRun = latestRun.items.find( - (candidate) => candidate.status === "running", - ); + const activeRun = yield* findAbandonedRun; if (activeRun === undefined) { return yield* error; }