diff --git a/packages/provider-cursor/src/server/CursorAgentSdk.test.ts b/packages/provider-cursor/src/server/CursorAgentSdk.test.ts index 4cc66b749d8a2..361e4fa76de63 100644 --- a/packages/provider-cursor/src/server/CursorAgentSdk.test.ts +++ b/packages/provider-cursor/src/server/CursorAgentSdk.test.ts @@ -3,6 +3,9 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, describe, it } from "@effect/vitest"; import { ProviderSessionId, ThreadId } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; +import * as Deferred from "effect/Deferred"; +import * as Fiber from "effect/Fiber"; +import * as TestClock from "effect/testing/TestClock"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Exit from "effect/Exit"; @@ -19,6 +22,7 @@ const cursorSdkMock = vi.hoisted(() => { let agentCloseFailure: unknown; // The local store lists runs oldest first, like the real SDK. let storedRuns: Array<{ readonly id: string; readonly status: string }> = []; + let onDelta: ((input: { readonly update: unknown }) => Promise) | undefined; const runWait = vi.fn(async () => ({ id: "run-cursor-agent-sdk-test", requestId: "request-cursor-agent-sdk-test", @@ -40,6 +44,7 @@ const cursorSdkMock = vi.hoisted(() => { _message: unknown, options: { readonly onDelta?: (input: { readonly update: unknown }) => Promise }, ) => { + onDelta = options.onDelta; runOperations.push("run.start"); if (storedRuns.some((run) => run.status === "running")) { throw new Error("Agent agent-cursor-agent-sdk-test already has active run"); @@ -88,6 +93,7 @@ const cursorSdkMock = vi.hoisted(() => { return { agentClose, + emitDelta: (update: unknown) => onDelta?.({ update }), cancelRun, closeExecutionOrder, create, @@ -272,6 +278,40 @@ describe("CursorAgentSdkRunner", () => { ]); }).pipe(Effect.provide(layerTest)), ); + it.effect("drains queued callbacks before returning from cancellation", () => + Effect.gen(function* () { + const order: Array = []; + const callbackStarted = yield* Deferred.make(); + const runner = yield* CursorAgentSdk.CursorAgentSdkRunner; + const session = yield* runner.open({ + operation: "create", + options: { model: { id: "default" }, mode: "agent", local: { cwd: process.cwd() } }, + threadId: ThreadId.make("thread-cursor-cancel-callback"), + providerSessionId: ProviderSessionId.make("session-cursor-cancel-callback"), + }); + const run = yield* session.send({ + message: "hi", + onDelta: (update) => + Effect.gen(function* () { + if (update.type !== "text-delta") return; + yield* Deferred.succeed(callbackStarted, undefined); + yield* Effect.sleep("1 second"); + order.push(update.text); + }), + }); + cursorSdkMock.runCancel.mockImplementationOnce(async () => { + void cursorSdkMock.emitDelta({ type: "text-delta", text: "callback" }); + }); + const cancelling = yield* run.cancel.pipe( + Effect.tap(() => Effect.sync(() => order.push("cancel returned"))), + Effect.forkScoped, + ); + yield* Deferred.await(callbackStarted); + yield* TestClock.adjust("1 second"); + yield* Fiber.join(cancelling); + assert.deepStrictEqual(order, ["callback", "cancel returned"]); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); it.effect("forwards the agent's own shell stream output as shell-output-delta", () => Effect.gen(function* () { diff --git a/packages/provider-cursor/src/server/CursorAgentSdk.ts b/packages/provider-cursor/src/server/CursorAgentSdk.ts index ef6b609bb3f60..29652a6d1d3e5 100644 --- a/packages/provider-cursor/src/server/CursorAgentSdk.ts +++ b/packages/provider-cursor/src/server/CursorAgentSdk.ts @@ -706,6 +706,7 @@ export const makeCursorAgentSdkRunner = Effect.fn("makeCursorAgentSdkRunner")(fu throw cause; } } + await callbackChain; }, catch: (cause) => runnerError(cause, "run.cancel"), }), diff --git a/packages/provider-cursor/src/server/adapter.test.ts b/packages/provider-cursor/src/server/adapter.test.ts index 1e1e17a1a0ec0..d515b48cb0f0c 100644 --- a/packages/provider-cursor/src/server/adapter.test.ts +++ b/packages/provider-cursor/src/server/adapter.test.ts @@ -25,10 +25,12 @@ import * as ChildProcess from "effect/process/ChildProcess"; import { spawnAndCollect } from "@t3tools/provider-core/server/snapshotProbe"; import * as HostProcess from "@t3tools/shared/HostProcess"; import * as FileSystem from "effect/FileSystem"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { vi } from "vite-plus/test"; import { symlinksSupported } from "@t3tools/shared/testing/symlinks"; @@ -403,6 +405,118 @@ describe("CursorAdapterV2", () => { message: "Agent turn stopped after repeated resume attempts made no progress", }, }, + { + reason: "turn-ended when run.wait never resolves", + cancelCalls: 1, + updates: [ + { type: "text-delta", text: "Done." }, + { type: "turn-ended" }, + ] satisfies ReadonlyArray, + wait: Effect.never, + failure: null, + }, + { + reason: "turn-ended while a tool is still running until tool-call-completed", + updates: [ + { type: "text-delta", text: "Checking." }, + { + type: "tool-call-started", + modelCallId: "native-model-call", + callId: "read-file", + toolCall: { + type: "read", + args: { path: "src/env.ts" }, + }, + }, + { type: "turn-ended" }, + { + type: "tool-call-completed", + modelCallId: "native-model-call", + callId: "read-file", + toolCall: { + type: "read", + args: { path: "src/env.ts" }, + result: { + status: "success", + value: { fileSize: 12, content: "ok", totalLines: 1 }, + }, + }, + }, + ] satisfies ReadonlyArray, + wait: Effect.never, + failure: null, + }, + { + reason: "a transport failure when run.wait never resolves after turn-ended", + updates: [ + { type: "text-delta", text: "Error: RetriableError: WritableIterable is closed" }, + { type: "turn-ended" }, + ] satisfies ReadonlyArray, + wait: Effect.never, + failure: { class: "transport_error", message: "Provider turn failed." }, + }, + { + reason: "the SDK fallback answer after turn-ended", + updates: [{ type: "turn-ended" }] satisfies ReadonlyArray, + wait: Effect.succeed({ + id: "native-cursor-run", + status: "finished" as const, + result: "The answer.", + }), + failure: null, + answers: ["The answer."], + cancelCalls: 0, + }, + { + reason: "a cancelled wait released by idle cancellation", + updates: [ + { type: "text-delta", text: "Done." }, + { type: "turn-ended" }, + ] satisfies ReadonlyArray, + wait: null, + failure: null, + cancelCalls: 1, + }, + { + reason: "a stop-hook continuation after turn-ended", + updates: [ + { type: "text-delta", text: "Checking." }, + { type: "turn-ended" }, + ] satisfies ReadonlyArray, + continuation: [ + { type: "text-delta", text: "Done." }, + { type: "turn-ended" }, + ] satisfies ReadonlyArray, + wait: null, + failure: null, + cancelCalls: 1, + answers: ["Checking.", "Done."], + }, + { + reason: "an interruption after idle cancellation timed out with a hung wait", + updates: [ + { type: "text-delta", text: "Done." }, + { type: "turn-ended" }, + ] satisfies ReadonlyArray, + wait: Effect.never, + failure: null, + status: "interrupted", + cancelOutcome: "timeout" as const, + interrupt: true, + cancelCalls: 2, + }, + { + reason: "a cancelled wait when idle cancel then fails", + updates: [ + { type: "text-delta", text: "Done." }, + { type: "turn-ended" }, + ] satisfies ReadonlyArray, + wait: null, + failure: null, + status: "cancelled", + cancelCalls: 1, + cancelOutcome: "after-wait" as const, + }, { reason: "no answer from a resume after the turn ended that made no progress", updates: [{ type: "turn-ended" }] satisfies ReadonlyArray, @@ -418,8 +532,13 @@ describe("CursorAdapterV2", () => { failure: null, answers: [], }, - ])("ends the turn with $reason", ({ updates = [], wait, failure, answers }) => + ])("ends the turn with $reason", (testCase) => Effect.gen(function* () { + const { updates = [], wait, failure, status, cancelCalls } = testCase; + const answers = "answers" in testCase ? testCase.answers : undefined; + const continuation = "continuation" in testCase ? testCase.continuation : undefined; + const interrupt = "interrupt" in testCase && testCase.interrupt; + const cancelOutcome = "cancelOutcome" in testCase ? testCase.cancelOutcome : undefined; const fileSystem = yield* FileSystem.FileSystem; const workspace = yield* fileSystem.makeTempDirectoryScoped({ prefix: "cursor-v2-failure-" }); const instanceId = ProviderInstanceId.make("cursor"); @@ -430,6 +549,10 @@ describe("CursorAdapterV2", () => { interactionMode: "default", cwd: workspace, }); + const cancelled = yield* Deferred.make(); + const waitObserved = yield* Deferred.make(); + let observedCancelCalls = 0; + let onDelta: ((update: InteractionUpdate) => Effect.Effect) | undefined; const adapter = yield* makeCursorAdapterV2({ instanceId, environment: { HOME: workspace }, @@ -443,14 +566,35 @@ describe("CursorAdapterV2", () => { close: Effect.void, send: (input) => Effect.gen(function* () { + onDelta = (update) => input.onDelta!(update).pipe(Effect.orDie); for (const update of updates) { yield* input.onDelta!(update).pipe(Effect.orDie); } return { agentId: "native-cursor-failure", runId: "native-cursor-run", - wait, - cancel: Effect.void, + wait: + wait ?? + Deferred.await(cancelled).pipe( + Effect.andThen(Deferred.succeed(waitObserved, undefined)), + Effect.as({ + id: "native-cursor-run", + status: "cancelled" as const, + }), + ), + cancel: Effect.gen(function* () { + observedCancelCalls += 1; + if (observedCancelCalls > 1) return; + yield* Deferred.succeed(cancelled, undefined); + if (cancelOutcome === "after-wait") yield* Deferred.await(waitObserved); + if (cancelOutcome === "timeout") return yield* Effect.never; + if (cancelOutcome !== undefined) { + return yield* new CursorAgentSdk.CursorAgentSdkRunnerError({ + method: "run.cancel", + cause: new Error("cancel failed"), + }); + } + }), }; }), }), @@ -510,18 +654,45 @@ describe("CursorAdapterV2", () => { attachments: [], }, }); + if (continuation !== undefined) { + yield* TestClock.adjust("90 seconds"); + assert.equal(observedCancelCalls, 0); + for (const update of continuation) { + yield* onDelta!(update); + } + } + yield* TestClock.adjust("2 minutes"); + if (interrupt) { + const started = yield* runtime.events.pipe( + Stream.filter((event) => event.type === "provider_turn.updated"), + Stream.take(1), + Stream.runCollect, + ); + const interrupting = yield* runtime + .interruptTurn({ + providerThread, + providerTurnId: started[0]!.providerTurn.id, + }) + .pipe(Effect.forkScoped); + yield* TestClock.adjust("10 seconds"); + yield* Fiber.join(interrupting); + } const events = yield* runtime.events.pipe( Stream.takeUntil((event) => event.type === "turn.terminal"), Stream.runCollect, ); const terminal = events.find((event) => event.type === "turn.terminal"); - assert.equal(terminal?.status, failure === null ? "completed" : "failed"); + assert.equal(terminal?.status, status ?? (failure === null ? "completed" : "failed")); + if (cancelCalls !== undefined) assert.equal(observedCancelCalls, cancelCalls); + if (cancelOutcome !== undefined) assert.equal(terminal?.threadDisposition, "broken"); assert.equal(terminal?.failure?.class, failure?.class); assert.equal(terminal?.failure?.message, failure?.message); if (answers !== undefined) { assert.deepEqual( events.flatMap((event) => - event.type === "message.updated" && event.message.role === "assistant" + event.type === "message.updated" && + event.message.role === "assistant" && + !event.message.streaming ? [event.message.text] : [], ), @@ -2308,6 +2479,7 @@ describe("CursorAdapterV2", () => { const steered: string[] = []; let cancelled = 0; let outcome: "complete_delivered" | "revert_to_followup" = "complete_delivered"; + const acknowledged = yield* Deferred.make(); const finished = yield* Deferred.make<{ readonly id: string; readonly requestId: string; @@ -2334,30 +2506,36 @@ describe("CursorAdapterV2", () => { agentId: "native-cursor-steer", listMessages: Effect.succeed([]), close: Effect.void, - send: () => - Effect.succeed({ - agentId: "native-cursor-steer", - runId: "native-cursor-steer-run", - wait: Deferred.await(finished), - cancel: Effect.sync(() => { - cancelled += 1; - }).pipe( - Effect.andThen( - Deferred.succeed(finished, { - id: "native-cursor-steer-run", - requestId: "native-request", - status: "cancelled" as const, - model: { id: "composer-2.5" }, - durationMs: 1, - }), + send: (input) => + Effect.gen(function* () { + yield* input.onDelta!({ type: "text-delta", text: "Done." }); + yield* input.onDelta!({ type: "turn-ended" }); + }).pipe( + Effect.orDie, + Effect.as({ + agentId: "native-cursor-steer", + runId: "native-cursor-steer-run", + wait: Deferred.await(finished), + cancel: Effect.sync(() => { + cancelled += 1; + }).pipe( + Effect.andThen( + Deferred.succeed(finished, { + id: "native-cursor-steer-run", + requestId: "native-request", + status: "cancelled" as const, + model: { id: "composer-2.5" }, + durationMs: 1, + }), + ), ), - ), - steer: (text) => - Effect.sync(() => { - steered.push(text); - return outcome; - }), - }), + steer: (text) => + Effect.sync(() => steered.push(text)).pipe( + Effect.andThen(Deferred.await(acknowledged)), + Effect.as(outcome), + ), + }), + ), }), }), Effect.provide(TestProviderHost.layer({ cwd: workspace })), @@ -2424,20 +2602,29 @@ describe("CursorAdapterV2", () => { nativeTurnId: "native-cursor-steer-run", }); yield* runtime.startTurn(turn); - yield* runtime.steerTurn({ - threadId, - runId: turn.runId, - providerThread, - providerTurnId, - message: { - messageId: MessageId.make("cursor-steer-follow-up"), - createdBy: "user", - creationSource: "web", - text: "$review Look at the tests instead.", - attachments: [], - }, - }); + const steering = yield* runtime + .steerTurn({ + threadId, + runId: turn.runId, + providerThread, + providerTurnId, + message: { + messageId: MessageId.make("cursor-steer-follow-up"), + createdBy: "user", + creationSource: "web", + text: "$review Look at the tests instead.", + attachments: [], + }, + }) + .pipe(Effect.forkScoped); + // An acknowledgment that outlasts the idle window must not lose the run to idle cancel. + yield* TestClock.adjust("2 minutes"); + assert.equal(cancelled, 0); + yield* Deferred.succeed(acknowledged, undefined); + yield* Fiber.join(steering); assert.deepEqual(steered, ["/review Look at the tests instead."]); + // The delivered steer is new work, so the ended turn is not cancelled as idle. + yield* TestClock.adjust("2 minutes"); assert.equal(cancelled, 0); yield* runtime.steerTurn({ threadId, diff --git a/packages/provider-cursor/src/server/adapter.ts b/packages/provider-cursor/src/server/adapter.ts index 1e160be188f08..fd844923c3c58 100644 --- a/packages/provider-cursor/src/server/adapter.ts +++ b/packages/provider-cursor/src/server/adapter.ts @@ -34,6 +34,7 @@ import { import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; @@ -963,6 +964,10 @@ interface ActiveCursorTurn { finalized: boolean; /** The agent's latest turn ended and no new work has started since. */ turnEnded: boolean; + /** Bumps when new work starts so a pending idle finalize is ignored. */ + idleFinalizeGeneration: number; + idleCancelResult: Deferred.Deferred | null; + steersInFlight: number; } interface CursorLiveAgent { @@ -2028,6 +2033,7 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( update.type === "tool-call-started" ) { context.turnEnded = false; + context.idleFinalizeGeneration += 1; } switch (update.type) { case "text-delta": @@ -2065,11 +2071,17 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( yield* completeAssistant(context); yield* completeReasoning(context); yield* handleToolUpdate(context, update); + if (update.type === "tool-call-completed") { + yield* scheduleIdleTurnFinalize(context); + } return; case "step-completed": case "turn-ended": yield* completeAssistant(context); yield* completeReasoning(context); + if (update.type === "turn-ended") { + yield* scheduleIdleTurnFinalize(context); + } return; case "shell-output-delta": { const toolCallId = update.event.toolCallId; @@ -2223,6 +2235,79 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( yield* Deferred.succeed(input.context.completed, undefined); }); + const cursorTurnHasActiveWork = (context: ActiveCursorTurn) => + context.steersInFlight > 0 || + context.tools.size > 0 || + Array.from(context.subagents.values()).some((subagent) => + isOrchestrationV2WorkActive(subagent.task.status), + ); + + const scheduleIdleTurnFinalize = Effect.fnUntraced(function* (context: ActiveCursorTurn) { + if ( + context.finalized || + !context.turnEnded || + context.assistant.nextSegment === 0 || + context.idleCancelResult !== null || + cursorTurnHasActiveWork(context) + ) { + return; + } + const generation = ++context.idleFinalizeGeneration; + yield* Effect.forkIn( + Effect.gen(function* () { + // Stop hooks run after turn-ended (in parallel, 60s default timeout) and a follow-up + // continues the run; wait past them before treating a quiet run as hung. + yield* Effect.sleep("2 minutes"); + if ( + context.finalized || + context.interrupted || + context.idleFinalizeGeneration !== generation || + !context.turnEnded || + cursorTurnHasActiveWork(context) + ) { + return; + } + if (context.idleCancelResult !== null) return; + const cancelResult = Deferred.makeUnsafe(); + context.idleCancelResult = cancelResult; + // Keep ownership if cancellation fails; a terminal wait still reports its own outcome. + const released = yield* context.run.cancel.pipe( + Effect.timeout("10 seconds"), + Effect.as(true), + Effect.catch((cause) => + Effect.logWarning("orchestration-v2.cursor-idle-cancel-failed", { + providerTurnId: context.providerTurnId, + cause, + }).pipe(Effect.as(false)), + ), + ); + yield* Deferred.succeed(cancelResult, released); + if (!released || context.finalized) return; + const transportFailure = context.assistantReply.failure; + const status = context.interrupted + ? "interrupted" + : transportFailure !== undefined + ? "failed" + : context.turnEnded && !cursorTurnHasActiveWork(context) + ? "completed" + : "cancelled"; + yield* finalizeTurn({ + context, + status, + ...(status === "failed" + ? { + failure: makeProviderFailure({ + cause: transportFailure, + class: "transport_error", + }), + } + : {}), + }); + }), + sessionScope, + ); + }); + const isResumeFailureAfterTurnEnded = (context: ActiveCursorTurn, result: RunResult) => context.turnEnded && result.status === "error" && @@ -2538,6 +2623,9 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( interrupted: false, finalized: false, turnEnded: false, + idleFinalizeGeneration: 0, + idleCancelResult: null, + steersInFlight: 0, }; yield* Ref.set(activeTurn, context); yield* emitProviderEvent({ @@ -2595,12 +2683,28 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( ); } if (context !== null) { + const idleCancel = + result.status === "cancelled" && context.idleCancelResult !== null + ? yield* Deferred.await(context.idleCancelResult) + : false; const transportFailure = context.assistantReply.failure; const status = - transportFailure === undefined ? terminalStatus(context, result) : "failed"; + transportFailure !== undefined + ? "failed" + : idleCancel && + !context.interrupted && + context.turnEnded && + !cursorTurnHasActiveWork(context) + ? "completed" + : terminalStatus(context, result); yield* finalizeTurn({ context, status, + ...(result.status === "cancelled" && + context.idleCancelResult !== null && + !idleCancel + ? { threadDisposition: "broken" as const } + : {}), ...(status === "failed" ? { failure: @@ -2783,26 +2887,43 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( detail: `Cursor provider turn ${turnInput.providerTurnId} is not active.`, }); } - // The live run already has the runtime instructions from startTurn. - const text = yield* providerMessageTextWithAttachmentPaths({ - text: yield* resolveCursorSkillText( - turnInput.message.text, - context.input.runtimePolicy.cwd ?? undefined, - ), - attachments: turnInput.message.attachments, - resolveAttachmentPath: host.resolveAttachmentPath, - resolveExtractedTextPath: host.resolveExtractedTextPath, - }); - if (text.trim().length === 0) { - return yield* new ProviderAdapter.ProviderAdapterProtocolError({ - driver: CursorAgentSdk.CURSOR_PROVIDER, - detail: "Cursor steer requires non-empty text.", + // Hold idle finalize from the moment a steer arrives until the run acknowledges it. + context.steersInFlight += 1; + context.idleFinalizeGeneration += 1; + const outcome = yield* Effect.gen(function* () { + // The live run already has the runtime instructions from startTurn. + const text = yield* providerMessageTextWithAttachmentPaths({ + text: yield* resolveCursorSkillText( + turnInput.message.text, + context.input.runtimePolicy.cwd ?? undefined, + ), + attachments: turnInput.message.attachments, + resolveAttachmentPath: host.resolveAttachmentPath, + resolveExtractedTextPath: host.resolveExtractedTextPath, }); - } - const outcome = - context.run.steer === undefined + if (text.trim().length === 0) { + return yield* new ProviderAdapter.ProviderAdapterProtocolError({ + driver: CursorAgentSdk.CURSOR_PROVIDER, + detail: "Cursor steer requires non-empty text.", + }); + } + // A run being cancelled for idleness cannot take more input. + return context.run.steer === undefined || context.idleCancelResult !== null ? "revert_to_followup" : yield* context.run.steer(text); + }).pipe( + Effect.onExit((exit) => + Effect.suspend(() => { + context.steersInFlight -= 1; + if (Exit.isSuccess(exit) && exit.value === "complete_delivered") { + context.turnEnded = false; + context.idleFinalizeGeneration += 1; + return Effect.void; + } + return scheduleIdleTurnFinalize(context); + }), + ), + ); if (outcome === "complete_delivered") { return; } @@ -2855,7 +2976,13 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( providerThreadId: turnInput.providerThread.id, providerTurnId: turnInput.providerTurnId, }); - yield* finalizeTurn({ context, status: "interrupted" }); + yield* finalizeTurn({ + context, + status: "interrupted", + ...(context.idleCancelResult !== null + ? { threadDisposition: "broken" as const } + : {}), + }); }, (effect, turnInput) => effect.pipe(