fix(provider-cursor): finalize turn when SDK wait hangs after turn-ended

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<void>) | 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<void> },
     ) => {
+      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<string> = [];
+      const callbackStarted = yield* Deferred.make<void>();
+      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* () {

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"),
                 }),