diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 02d6f64c0be66..7e23c65625bb4 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -7502,6 +7502,92 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest)), ); + it.live("settles interrupts that arrive after the session scope has closed", () => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const path = yield* Path.Path; + const selfInvocation = yield* resolveSelfInvocation(); + const mockAgentPath = yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ); + const protocolEvents = yield* Queue.bounded(256); + const instanceId = ProviderInstanceId.make("acp-test"); + const adapter = yield* makeAcpAdapterV2({ + instanceId, + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + interruptPromptOnCancel: false, + restartRuntimeAfterInterrupt: true, + terminateRuntimeProcessGroupOnInterrupt: true, + makeRuntime: makeMockRuntime({ + childProcessSpawner, + mockAgentPath, + ownDetachedProcessGroup: true, + environment: { T3_ACP_HANG_PROMPT_FOREVER: "1" }, + protocolEvents, + }), + }, + selfInvocation, + }); + const threadId = ThreadId.make("thread-acp-interrupt-after-scope-close"); + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }); + const modelSelection = { instanceId, model: "default" } as const; + const sessionScope = yield* Scope.make(); + const runtime = yield* adapter + .openSession({ + threadId, + providerSessionId: ProviderSessionId.make( + "provider-session-acp-interrupt-after-scope-close", + ), + modelSelection, + runtimePolicy, + }) + .pipe(Effect.provideService(Scope.Scope, sessionScope)); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const now = yield* DateTime.now; + yield* runtime + .startTurn(makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now })) + .pipe(Effect.forkScoped); + yield* Stream.fromQueue(protocolEvents).pipe( + Stream.filter( + (event) => + event.direction === "outgoing" && rawProtocolMethod(event) === "session/prompt", + ), + Stream.runHead, + ); + const providerTurnId = idAllocator.derive.providerTurn({ + driver: ACP_TEST_DRIVER, + nativeTurnId: acpScopedNativeId(instanceId, "mock-session-1:turn:1"), + }); + // A worktree hand-off releases the session, then ingestion cleanup interrupts again. + yield* Scope.close(sessionScope, Exit.void); + const detachInterrupt = yield* Effect.exit( + runtime.interruptTurn({ providerThread, providerTurnId }), + ); + assert.isFalse( + Exit.isFailure(detachInterrupt) && Cause.hasInterruptsOnly(detachInterrupt.cause), + "an interrupt nobody cancelled must not exit as interrupted", + ); + const cleanupInterrupt = yield* Effect.exit( + runtime.interruptTurn({ providerThread, providerTurnId, requestRuntimeRestart: true }), + ); + assert.isFalse( + Exit.isFailure(cleanupInterrupt) && Cause.hasInterruptsOnly(cleanupInterrupt.cause), + "a later interrupt must settle instead of waiting on a leaked transition permit", + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.effect( "carries a live subagent lineage across an interrupt so the next turn can complete it", () => diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 24cde9fd967ba..4155c9fc9f4c4 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -9410,7 +9410,13 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ); }).pipe( Effect.ensuring(runtimeTransitionPermit.release(1)), - Effect.forkIn(sessionScope, { startImmediately: true, uninterruptible: true }), + // forkIn on a closed scope never starts the fiber, which would leak the permit. + sessionScope.state._tag === "Closed" + ? Effect.forkDetach({ startImmediately: true, uninterruptible: true }) + : Effect.forkIn(sessionScope, { + startImmediately: true, + uninterruptible: true, + }), ); return yield* restore(Fiber.join(transitionFiber)).pipe( Effect.onInterrupt(() =>