From 31fab0f420521d38f1695a3d61309523d4e93267 Mon Sep 17 00:00:00 2001
From: Trevor Edwards <9443483+trevoedwards@users.noreply.github.com>
Date: Mon, 2 Nov 2026 00:45:36 +0000
Subject: [PATCH] fix(provider-acp): replacement ACP runtimes keep their event
 consumer

When rollback or a forced new native thread replaced the ACP runtime, the
replacement's event consumer was forked into the previous runtime's scope,
which the same transition then closed. Nothing read the replacement's events
afterwards: onSessionEvent stopped, the event queue grew unbounded, and
wait-for-prompt cancellation waited on an unacknowledged drain barrier until
its timeout killed the agent. Fork the consumer into the replacement scope.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../orchestration-v2/Adapters/AcpAdapterV2.test.ts  | 13 +++++++++++++
 packages/provider-acp/src/server/adapter.ts         | 10 ++++++++--
 2 files changed, 21 insertions(+), 2 deletions(-)

diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts
index bb84c1a2c7492..db0d7f9c1eea5 100644
--- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts
+++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts
@@ -4297,10 +4297,16 @@ describe("AcpAdapterV2", () => {
         browserToolsAvailable: false,
       });
       const runtimeInputs: Array<AcpAdapterV2RuntimeInput> = [];
+      const nativeRuntimes: Array<AcpSessionRuntime.AcpSessionRuntime["Service"]> = [];
+      let sessionEventCount = 0;
       const makeRuntime = makeMockRuntime({
         childProcessSpawner,
         mockAgentPath,
         environment: { T3_ACP_PROMPT_DELAY_MS: "100" },
+        wrapRuntime: (nativeRuntime) => {
+          nativeRuntimes.push(nativeRuntime);
+          return nativeRuntime;
+        },
       });
       const adapter = yield* makeAcpAdapterV2({
         instanceId,
@@ -4311,6 +4317,10 @@ describe("AcpAdapterV2", () => {
             Effect.sync(() => runtimeInputs.push(runtimeInput)).pipe(
               Effect.andThen(makeRuntime(runtimeInput)),
             ),
+          onSessionEvent: () =>
+            Effect.sync(() => {
+              sessionEventCount += 1;
+            }),
         },
         selfInvocation,
       });
@@ -4405,6 +4415,7 @@ describe("AcpAdapterV2", () => {
       });
       assert.deepEqual(snapshotAfterRollback.providerTurns, []);
       assert.deepEqual(snapshotAfterRollback.messages, []);
+      const sessionEventsBeforeSecondTurn = sessionEventCount;
 
       yield* runtime.startTurn(
         makeTurnInput({
@@ -4431,6 +4442,8 @@ describe("AcpAdapterV2", () => {
         }
       }
       assert.equal(secondStatus, "completed");
+      yield* nativeRuntimes[1]!.drainEvents;
+      assert.isAbove(sessionEventCount, sessionEventsBeforeSecondTurn);
     }).pipe(Effect.provide(layerTest), Effect.scoped),
   );
 
diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts
index 7b24d9ef09adf..78e354ae00d50 100644
--- a/packages/provider-acp/src/server/adapter.ts
+++ b/packages/provider-acp/src/server/adapter.ts
@@ -6047,6 +6047,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
             readonly sessionUpdates?: boolean;
             readonly terminals?: boolean;
             readonly mcp?: boolean;
+            readonly scope?: Scope.Scope;
           } = {},
         ) {
           yield* targetRuntime.getEvents().pipe(
@@ -6059,7 +6060,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
                 flavor.onSessionEvent?.(event) ?? Effect.void,
               ).pipe(Effect.asVoid);
             }),
-            Effect.forkIn(runtimeScope ?? sessionScope),
+            Effect.forkIn(handlerOptions.scope ?? runtimeScope ?? sessionScope),
           );
           const requestUserInput = (
             request: AcpAdapterV2UserInputRequest,
@@ -6607,7 +6608,12 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
                   yield* wireAcpRuntimeHandlers(
                     replacementExit.value.replacementRuntime,
                     replacementGeneration,
-                    { sessionUpdates: false, terminals: false, mcp: false },
+                    {
+                      sessionUpdates: false,
+                      terminals: false,
+                      mcp: false,
+                      scope: replacementScope,
+                    },
                   );
                   const lifecycle = yield* Ref.modify(
                     candidateLifecycle,