Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 66 additions & 7 deletions packages/provider-cursor/src/server/CursorAgentSdk.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ const cursorSdkMock = vi.hoisted(() => {
const runOperations: Array<string> = [];
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",
Expand Down Expand Up @@ -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;
Expand All @@ -87,6 +111,9 @@ const cursorSdkMock = vi.hoisted(() => {
setAgentCloseFailure: (failure: unknown) => {
agentCloseFailure = failure;
},
setListRunsPageSize: (size: number | undefined) => {
listRunsPageSize = size;
},
setStaleRunActive: (active: boolean) => {
staleRunActive = active;
},
Expand Down Expand Up @@ -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",
Expand All @@ -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)),
);
});
65 changes: 58 additions & 7 deletions packages/provider-cursor/src/server/CursorAgentSdk.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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<object>();
Expand Down Expand Up @@ -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;
}
Expand Down
Loading