fix(acp): recover sessions after cancellation fails or stalls

apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts

@@ -12222,6 +12222,152 @@ describe("AcpAdapterV2", () => {
       }).pipe(Effect.provide(layerTest), Effect.scoped),
   );
 
+  it.effect.each([
+    ["failed", "resume"],
+    ["unacknowledged", "resume"],
+    ["failed", "load-only"],
+  ] as const)(
+    "recovers from %s cancellation with %s support on a fresh runtime",
+    ([cancellation, sessionSupport]) =>
+      Effect.gen(function* () {
+        const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner;
+        const path = yield* Path.Path;
+        const idAllocator = yield* IdAllocator.IdAllocatorV2;
+        const mockAgentPath = yield* path.fromFileUrl(
+          new URL("../../../scripts/acp-mock-agent.ts", import.meta.url),
+        );
+        const protocolEvents = yield* Queue.unbounded<EffectAcpProtocol.AcpProtocolLogEvent>();
+        const promptStarted = yield* Deferred.make<void>();
+        const cancelAttempted = yield* Deferred.make<void>();
+        const instanceId = ProviderInstanceId.make("acp-cancel-failure");
+        let runtimeCount = 0;
+        const makeRuntime = makeMockRuntime({
+          childProcessSpawner,
+          mockAgentPath,
+          protocolEvents,
+          wrapRuntime: (runtime, ordinal) => {
+            runtimeCount = ordinal;
+            const sessionRuntime =
+              sessionSupport === "load-only"
+                ? {
+                    ...runtime,
+                    start: () =>
+                      runtime.start().pipe(
+                        Effect.map((started) => ({
+                          ...started,
+                          initializeResult: {
+                            ...started.initializeResult,
+                            agentCapabilities: {
+                              ...started.initializeResult.agentCapabilities,
+                              loadSession: true,
+                              sessionCapabilities: {},
+                            },
+                          },
+                        })),
+                      ),
+                  }
+                : runtime;
+            return ordinal === 1
+              ? {
+                  ...sessionRuntime,
+                  prompt: () =>
+                    Deferred.succeed(promptStarted, undefined).pipe(Effect.andThen(Effect.never)),
+                  cancel: Deferred.succeed(cancelAttempted, undefined).pipe(
+                    Effect.andThen(
+                      cancellation === "failed"
+                        ? Effect.fail(
+                            EffectAcpErrors.AcpRequestError.internalError("cancel failed"),
+                          )
+                        : Effect.void,
+                    ),
+                  ),
+                }
+              : sessionRuntime;
+          },
+        });
+        const adapter = yield* makeAcpAdapterV2({
+          instanceId,
+          selfInvocation: yield* resolveSelfInvocation(),
+          flavor: {
+            driver: ACP_TEST_DRIVER,
+            capabilities: AcpProviderCapabilitiesV2,
+            preferResumeSession: true,
+            makeRuntime: (input) => {
+              const runtimeOptions = { ...input, resumeMethod: "resume" as const };
+              return makeRuntime(runtimeOptions);
+            },
+          },
+        });
+        const threadId = ThreadId.make("thread-acp-cancel-failure");
+        const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({
+          runtimeMode: "full-access",
+          interactionMode: "default",
+          cwd: process.cwd(),
+        });
+        const modelSelection = { instanceId, model: "default" } as const;
+        const session = yield* adapter.openSession({
+          threadId,
+          providerSessionId: ProviderSessionId.make("session-acp-cancel-failure"),
+          modelSelection,
+          runtimePolicy,
+        });
+        const providerThread = yield* session.ensureThread({
+          threadId,
+          modelSelection,
+          runtimePolicy,
+        });
+        const now = yield* DateTime.now;
+        yield* session.startTurn(
+          makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now }),
+        );
+        yield* Deferred.await(promptStarted);
+        const providerTurnId = idAllocator.derive.providerTurn({
+          driver: ACP_TEST_DRIVER,
+          nativeTurnId: acpScopedNativeId(
+            instanceId,
+            `${providerThread.nativeThreadRef?.nativeId}:turn:1`,
+          ),
+        });
+        const interrupt = yield* session
+          .interruptTurn({ providerThread, providerTurnId })
+          .pipe(Effect.exit, Effect.forkScoped);
+        yield* Deferred.await(cancelAttempted);
+        yield* TestClock.adjust(cancellation === "failed" ? "2 seconds" : "10 seconds");
+        const result = yield* Fiber.join(interrupt);
+        assert.isTrue(Exit.isFailure(result));
+        if (Exit.isFailure(result)) {
+          assert.include(
+            Cause.pretty(result.cause),
+            cancellation === "failed" ? "cancel failed" : "did not acknowledge cancellation",
+          );
+        }
+        yield* Queue.takeAll(protocolEvents);
+        yield* session.startTurn(
+          makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now, ordinal: 2 }),
+        );
+        assert.equal(runtimeCount, 2);
+        const terminals = yield* session.events.pipe(
+          Stream.takeUntil(
+            (event) => event.type === "turn.terminal" && event.providerTurnId !== providerTurnId,
+          ),
+          Stream.runCollect,
+        );
+        assert.deepEqual(
+          terminals.filter((event) => event.type === "turn.terminal").map((event) => event.status),
+          ["interrupted", "completed"],
+        );
+        const methods = (yield* Queue.takeAll(protocolEvents))
+          .filter((event) => event.direction === "outgoing")
+          .map(rawProtocolMethod);
+        if (sessionSupport === "load-only") {
+          assert.include(methods, "session/new");
+        } else {
+          assert.notInclude(methods, "session/new");
+        }
+        assert.equal(methods.filter((method) => method === "session/resume").length, 1);
+      }).pipe(Effect.provide(layerTest), Effect.scoped),
+  );
+
   it.effect("restarts the ACP child process before the next prompt after interrupt", () =>
     Effect.gen(function* () {
       const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner;

apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts

@@ -166,6 +166,7 @@ export function makeAntigravityAcpAdapterFlavor(
     // Loading history replays every tool call; resume restores the session
     // without the replay and is what the official client does.
     preferResumeSession: true,
+    restartRuntimeAfterInterrupt: true,
     subagentsIdleOnTurnCompletion: true,
     ...(options.onSessionEvent === undefined ? {} : { onSessionEvent: options.onSessionEvent }),
     applyModelSelection: ({ runtime, modelSelection }) =>

packages/provider-acp/src/server/AcpSessionRuntime.ts

@@ -2432,6 +2432,13 @@ export const make = (
       resumeSession: (sessionId, activationOptions) =>
         start.pipe(
           Effect.flatMap((started) => {
+            if (
+              started.sessionId === sessionId &&
+              options.resumeSessionId === sessionId &&
+              options.resumeMethod === "resume"
+            ) {
+              return Effect.succeed(started);
+            }
             const requestPayload = {
               sessionId,
               cwd: options.cwd,
@@ -2442,9 +2449,8 @@ export const make = (
               "session/resume",
               requestPayload,
               acp.agent.resumeSession(requestPayload),
-            );
+            ).pipe(Effect.flatMap((response) => adoptSession(sessionId, response)));
           }),
-          Effect.flatMap((response) => adoptSession(sessionId, response)),
         ),
       forkSession: (sessionId, activationOptions) =>
         start.pipe(

packages/provider-acp/src/server/adapter.ts

@@ -5595,8 +5595,14 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
           return yield* flavor.withRuntimeStartup?.(startup) ?? startup;
         });
 
-        const restartAcpRuntime = Effect.fnUntraced(function* (threadId: ThreadId | null) {
-          yield* spawnAcpRuntime(threadId);
+        const restartAcpRuntime = Effect.fnUntraced(function* (
+          threadId: ThreadId | null,
+          resumeSessionId?: string,
+        ) {
+          yield* spawnAcpRuntime(
+            threadId,
+            flavor.preferResumeSession === true && canResumeSession ? resumeSessionId : undefined,
+          );
           yield* wireAcpRuntimeHandlers(runtime, yield* Ref.get(runtimeCallbackGeneration));
         });
 
@@ -6385,10 +6391,11 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
 
         const restartRuntimeAfterTeardownIfRequired = Effect.fnUntraced(function* (
           threadId: ThreadId | null,
+          resumeSessionId?: string,
         ) {
           const restartRequired = yield* Ref.get(runtimeRestartRequired);
           if (!restartRequired) return false;
-          yield* restartAcpRuntime(threadId);
+          yield* restartAcpRuntime(threadId, resumeSessionId);
           yield* Ref.set(runtimeRestartRequired, false);
           yield* Ref.set(activeSessionId, null);
           yield* Ref.set(activeSessionSetup, null);
@@ -6422,6 +6429,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
             const requestedSessionId = yield* nativeThreadId(driver, turnInput.providerThread);
             const restartAfterInterrupt = yield* restartRuntimeAfterTeardownIfRequired(
               turnInput.threadId,
+              requestedSessionId,
             );
             const needsSessionActivation =
               (yield* Ref.get(activeSessionId)) !== requestedSessionId || restartAfterInterrupt;
@@ -6904,6 +6912,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
                   });
                   const restartAfterInterrupt = yield* restartRuntimeAfterTeardownIfRequired(
                     threadInput.providerThread.appThreadId,
+                    sessionId,
                   ).pipe(Effect.tapError(() => restorePreviousItemIdentity));
                   if ((yield* Ref.get(activeSessionId)) !== sessionId || restartAfterInterrupt) {
                     yield* Ref.set(snapshot, {
@@ -7185,7 +7194,15 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
                         // acknowledge and would only threaten still-running
                         // background subagents.
                         if (!settledSoftInterrupt) {
-                          yield* runtime.cancel;
+                          const cancelExit = yield* Effect.exit(runtime.cancel);
+                          if (Exit.isFailure(cancelExit)) {
+                            yield* Ref.set(runtimeRestartRequired, true);
+                            yield* Deferred.await(context.completed).pipe(
+                              Effect.timeoutOption("2 seconds"),
+                            );
+                            yield* finalizeTurn(context, "interrupted");
+                            return yield* Effect.failCause(cancelExit.cause);
+                          }
                         }
                         if (restartRuntime && flavor.restartRuntimeAfterInterrupt === true) {
                           yield* Ref.set(runtimeRestartRequired, true);
@@ -7195,6 +7212,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
                         Effect.timeoutOption("10 seconds"),
                       );
                       if (Option.isNone(stopped)) {
+                        yield* Ref.set(runtimeRestartRequired, true);
                         if (!context.finalized) {
                           yield* finalizeTurn(context, "interrupted");
                         }