diff --git a/apps/server/src/device/LocalDeviceHost.ts b/apps/server/src/device/LocalDeviceHost.ts index 2e3cb3d0f71a489c02dbb0a471bbf930fbcc342f..65ddf6895742ecdd9eff0d2537981a4639d1d7ce 100644 --- a/apps/server/src/device/LocalDeviceHost.ts +++ b/apps/server/src/device/LocalDeviceHost.ts @@ -383,62 +383,63 @@ export const make = Effect.fn("LocalDeviceHost.make")(function* () { ); const origin = `http://127.0.0.1:${port}`; const scope = yield* Scope.make("sequential"); - const child = yield* spawner - .spawn( - ChildProcess.make( - nodePath, - [ - hubTool.entryPath, - "--port", - String(port), - "--host", - "127.0.0.1", - "--hide-sidebar", - "--hide-boot-device", - ], - { - detached: false, - shell: false, - stdout: "pipe", - stderr: "pipe", - env: yield* hubEnvironment(hostEnvironment).pipe( - Effect.provideService(FileSystem.FileSystem, fs), - Effect.provideService(HostProcessPlatform, hostPlatform), - ), - }, - ), - ) - .pipe( - Effect.provideService(Scope.Scope, scope), - Effect.mapError( - (cause) => - new DeviceHost.DeviceHostError({ - hostId, - step: "starting the device hub", - cause, - }), - ), - ); - const startedAtMillis = yield* Clock.currentTimeMillis; - const hub: HubProcess = { child, scope, origin, startedAtMillis, nodePath }; - yield* Effect.forkIn(observeHubOutput(hub), scope); - yield* waitForHttpReady({ - baseUrl: origin, - path: "/readyz", - timeoutMs: HUB_READY_TIMEOUT_MS, - makeError: (info) => - new DeviceHost.DeviceHostError({ - hostId, - step: "waiting for the device hub to answer", - cause: info.cause, - }), - }).pipe( - Effect.provideService(HttpClient.HttpClient, httpClient), - Effect.tapError(() => stopHub(hub)), - ); - yield* recordHub(hub, hubTool); - yield* Effect.logInfo("Device hub started", { pid: Number(child.pid), port }); - return hub; + // On failure or interrupt, from the spawn on: the hub is not recorded yet, + // so nothing else stops it. + return yield* Effect.gen(function* () { + const child = yield* spawner + .spawn( + ChildProcess.make( + nodePath, + [ + hubTool.entryPath, + "--port", + String(port), + "--host", + "127.0.0.1", + "--hide-sidebar", + "--hide-boot-device", + ], + { + detached: false, + shell: false, + stdout: "pipe", + stderr: "pipe", + env: yield* hubEnvironment(hostEnvironment).pipe( + Effect.provideService(FileSystem.FileSystem, fs), + Effect.provideService(HostProcessPlatform, hostPlatform), + ), + }, + ), + ) + .pipe( + Effect.provideService(Scope.Scope, scope), + Effect.mapError( + (cause) => + new DeviceHost.DeviceHostError({ + hostId, + step: "starting the device hub", + cause, + }), + ), + ); + const startedAtMillis = yield* Clock.currentTimeMillis; + const hub: HubProcess = { child, scope, origin, startedAtMillis, nodePath }; + yield* Effect.forkIn(observeHubOutput(hub), scope); + yield* waitForHttpReady({ + baseUrl: origin, + path: "/readyz", + timeoutMs: HUB_READY_TIMEOUT_MS, + makeError: (info) => + new DeviceHost.DeviceHostError({ + hostId, + step: "waiting for the device hub to answer", + cause: info.cause, + }), + }).pipe(Effect.provideService(HttpClient.HttpClient, httpClient)); + yield* recordHub(hub, hubTool); + yield* Effect.logInfo("Device hub started", { pid: Number(child.pid), port }); + return hub; + }).pipe(Effect.onError(() => Scope.close(scope, Exit.void).pipe(Effect.ignore))); }); const observeHubOutput = (hub: HubProcess) => @@ -622,25 +623,30 @@ export const make = Effect.fn("LocalDeviceHost.make")(function* () { ); yield* onPhase("starting"); const hub = yield* spawnHub(hubTool, nodePath); - yield* pruneLocalDeviceTools(config.baseDir, nodePath, "hub").pipe( - Effect.provideService(Path.Path, path), - Effect.provideService(ProcessRunner.ProcessRunner, runner), - Effect.ignore, - ); - const candidate = helperPaths(hubTool); - const [axExists, cliExists] = yield* Effect.all([ - fs.exists(candidate.serveSimAxSettings).pipe(Effect.orElseSucceed(() => false)), - fs.exists(candidate.serveSimCli).pipe(Effect.orElseSucceed(() => false)), - ]); - const next: RunningHost = { - hub, - agentDevice: null, - helpers: { - serveSimAxSettings: axExists ? candidate.serveSimAxSettings : null, - serveSimCli: cliExists ? candidate.serveSimCli : null, - }, - }; - yield* Ref.set(runningRef, next); + // Until the hub is in runningRef, nothing else stops it, so publishing it + // is part of the guarded step. + const next = yield* Effect.gen(function* () { + yield* pruneLocalDeviceTools(config.baseDir, nodePath, "hub").pipe( + Effect.provideService(Path.Path, path), + Effect.provideService(ProcessRunner.ProcessRunner, runner), + Effect.ignore, + ); + const candidate = helperPaths(hubTool); + const [axExists, cliExists] = yield* Effect.all([ + fs.exists(candidate.serveSimAxSettings).pipe(Effect.orElseSucceed(() => false)), + fs.exists(candidate.serveSimCli).pipe(Effect.orElseSucceed(() => false)), + ]); + const next: RunningHost = { + hub, + agentDevice: null, + helpers: { + serveSimAxSettings: axExists ? candidate.serveSimAxSettings : null, + serveSimCli: cliExists ? candidate.serveSimCli : null, + }, + }; + yield* Ref.set(runningRef, next); + return next; + }).pipe(Effect.onError(() => stopHub(hub))); yield* Ref.set(restartDelayRef, 0); yield* Effect.forkDetach(superviseHub(hub, hubTool)); return next; diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 9af1c0efde4bd9e9c0955685f8cda9aab2d96df5..81f9abc001a374d893866a936cc312e5e38b0910 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -7148,7 +7148,8 @@ export function makeClaudeAdapterV2( options: queryOptions, }) .pipe( - Effect.tapError(() => + // An interrupted open leaves the old process just as dead. + Effect.onError(() => // Same-native-thread replacement: the old process is already // dead, so its process-scoped roster is not authoritative. // First-ever failed open (no prior live query) must not invent diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 19517c3491161c849ef40636739570f8fb03ab27..d562f876df496259eb8390270c6baee7dedc7c8a 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -6174,8 +6174,10 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi yield* Ref.update(pendingRootTurns, (current) => new Map(current).set(threadId, turnInput), ); + // Cleared on interrupt too, as startTurn does: an interrupted start + // must not adopt the native turn that a late turn/started reports. yield* client.request("thread/compact/start", { threadId }).pipe( - Effect.tapError(() => + Effect.onError(() => Ref.update(pendingRootTurns, (current) => { const next = new Map(current); next.delete(threadId); diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 46792897e7893be373c9275d262fcbe6ff9919f9..7274de8c386f0615b6baad00f3b2adbc1f832e62 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -428,6 +428,25 @@ describe("OpenCode2 adapter", () => { }).pipe(Effect.scoped), ); + it.effect("ends a turn on the provider thread it started on", () => + Effect.gen(function* () { + const { runtime, thread } = yield* resumed([ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + event("session.execution.succeeded", { sessionID: SESSION }), + ]); + // A forked run starts on its own row for the same native session, while + // the adapter tracks the session under the id it minted for the fork. + const forkedRow = { + ...thread, + id: ProviderThreadId.make("provider-thread:opencode2-adapter:forked-run-row"), + }; + const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped); + yield* runtime.startTurn(turnInput(forkedRow)); + assert.equal((yield* Fiber.join(terminal))?.providerThreadId, forkedRow.id); + }).pipe(Effect.scoped), + ); + it.effect("ends the turn when its terminal event is one this build cannot decode", () => Effect.gen(function* () { const { runtime, thread } = yield* resumed([ diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts index 7716b75baaf9f4568f60d383d0a2a0e545fb8d3b..102eb721bb1a67ce4a4bcc3e19a55ebe3f0602c1 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts @@ -1786,7 +1786,10 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const base = { type: "turn.terminal" as const, driver, - providerThreadId: state.providerThread.id, + // The provider thread the turn started on. After a native fork the + // session's own thread has a fresh id, and the terminal must name the + // thread the start was recorded against. + providerThreadId: turn.providerTurn.providerThreadId, providerTurnId: turn.providerTurn.id, runOrdinal: turn.input.runOrdinal, threadDisposition, @@ -2825,12 +2828,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr * reconciles. Retried a few times; the caller fails everything after that. */ const reconnect = Effect.gen(function* () { + // Returned on any failure, including the interrupt a closing session + // sends a reconnect still in flight, so a spawned server can idle-stop. const scope = yield* Scope.make(); - const next = yield* borrow.pipe( - Effect.provideService(Scope.Scope, scope), - Effect.tapError(() => Scope.close(scope, Exit.void)), - ); - const stream = yield* next.events.pipe(Effect.tapError(() => Scope.close(scope, Exit.void))); + const { next, stream } = yield* Effect.gen(function* () { + const next = yield* borrow.pipe(Effect.provideService(Scope.Scope, scope)); + return { next, stream: yield* next.events }; + }).pipe(Effect.onError(() => Scope.close(scope, Exit.void))); const previous = currentScope; connection = next; client = next.client; diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.test.ts index 44632e02c66d56921597c798da83f5e55a3e6757..a8e28a377d1aeab338888899d072b0e8230c90a5 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.test.ts @@ -182,7 +182,10 @@ const makeOpenCodeRuntimeHarness = Effect.fn("makeOpenCodeRuntimeHarness")(funct runtimePolicy: policy, }); const now = yield* DateTime.now; - const startTurn = (text = "hello") => + const startTurn = ( + text = "hello", + startProviderThread: OrchestrationV2ProviderThread = providerThread, + ) => runtime.startTurn({ appThread: { id: threadId, @@ -213,7 +216,7 @@ const makeOpenCodeRuntimeHarness = Effect.fn("makeOpenCodeRuntimeHarness")(funct providerTurnOrdinal: 1, attemptId: RunAttemptId.make(`attempt-opencode-${suffix}`), rootNodeId: NodeId.make(`node-opencode-${suffix}`), - providerThread, + providerThread: startProviderThread, message: { createdBy: "user", creationSource: "web", @@ -1291,6 +1294,45 @@ describe("OpenCodeAdapterV2", () => { }).pipe(Effect.provide(IdAllocator.layer), Effect.scoped), ); + it.effect("ends a turn on the provider thread it started on", () => + Effect.gen(function* () { + const nativeEvents = asyncEventStream(); + const harness = yield* makeOpenCodeRuntimeHarness( + "forked-row", + "native-opencode-forked-row", + { + event: { + subscribe: async (_input: unknown, options: { signal?: AbortSignal }) => { + options.signal?.addEventListener("abort", () => nativeEvents.close(), { once: true }); + return { stream: nativeEvents.stream }; + }, + }, + session: { + create: async () => ({ + data: { id: "native-opencode-forked-row", time: { created: 1, updated: 1 } }, + }), + summarize: async () => ({ data: true }), + }, + }, + ); + // A forked run starts on its own row for the same native session, while + // the adapter tracks the session under the id it minted. + const forkedRow = { + ...harness.providerThread, + id: ProviderThreadId.make("provider-thread:opencode-test:forked-run-row"), + }; + yield* harness.startTurn("/compact", forkedRow); + const terminal = yield* harness.runtime.events.pipe( + Stream.filter( + (event): event is Extract => + event.type === "turn.terminal", + ), + Stream.runHead, + ); + assert.equal(Option.getOrUndefined(terminal)?.providerThreadId, forkedRow.id); + }).pipe(Effect.provide(IdAllocator.layer), Effect.scoped), + ); + it.effect("does not restore messages beyond OpenCode's persisted revert boundary", () => Effect.gen(function* () { const nativeEvents = asyncEventStream(); diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts index aad31d6363258c48c65e01a457b29d13688d868e..510a4c859d362e9f037fc3dbbe4bf9cdf15bfcd5 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts @@ -279,6 +279,8 @@ interface ActiveOpenCodeTurn { readonly modelSelection: ModelSelection; readonly runtimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy; readonly providerTurnId: OrchestrationV2ProviderTurn["id"]; + /** The provider thread the turn started on, which its terminal names. */ + readonly providerThreadId: OrchestrationV2ProviderTurn["providerThreadId"]; readonly providerTurnOrdinal: number; readonly runOrdinal: number; readonly runAttemptId: OrchestrationV2ProviderTurn["runAttemptId"]; @@ -2118,7 +2120,7 @@ export function makeOpenCodeAdapterV2( ? { type: "turn.terminal", driver: OPENCODE_PROVIDER, - providerThreadId: state.providerThread.id, + providerThreadId: turn.providerThreadId, providerTurnId: turn.providerTurnId, runOrdinal: turn.runOrdinal, failureItemOrdinal: itemOrdinal(turn, `terminal-failure:${turn.providerTurnId}`), @@ -2134,7 +2136,7 @@ export function makeOpenCodeAdapterV2( : { type: "turn.terminal", driver: OPENCODE_PROVIDER, - providerThreadId: state.providerThread.id, + providerThreadId: turn.providerThreadId, providerTurnId: turn.providerTurnId, runOrdinal: turn.runOrdinal, status, @@ -2256,6 +2258,7 @@ export function makeOpenCodeAdapterV2( modelSelection: state.appThread.modelSelection, runtimePolicy: state.parentSubagent.parentTurn.runtimePolicy, providerTurnId, + providerThreadId: providerTurn.providerThreadId, providerTurnOrdinal: providerTurn.ordinal, runOrdinal: state.parentSubagent.parentTurn.runOrdinal, runAttemptId: null, @@ -3192,6 +3195,7 @@ export function makeOpenCodeAdapterV2( modelSelection: turnInput.modelSelection, runtimePolicy: turnInput.runtimePolicy, providerTurnId, + providerThreadId: turnInput.providerThread.id, providerTurnOrdinal: turnInput.providerTurnOrdinal, runOrdinal: turnInput.runOrdinal, runAttemptId: turnInput.attemptId, diff --git a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts index 87e24f92b0e8d5191c8cbbeb6a744d4e1ab91260..9270253d5c74bc01a0f59ba42d1b9290bd0c8959 100644 --- a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts @@ -2679,8 +2679,9 @@ export function makePiAdapterV2( } // Pi fork replaces the session file, including for rollback. Persist // its new identity before any later request can fail or restart. + // An interrupted read leaves the identity just as unknown as a failed one. const forkState = yield* request({ type: "get_state" }).pipe( - Effect.tapError(() => + Effect.onError(() => Effect.sync(() => { threadState = null; }), diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 61a3dedafd53f99fbcb361bd8d8460ddeeb56bbd..bfb076ee3bf3b6d3f3048efa917574e38f2fb6bd 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -128,6 +128,30 @@ const layerFlakyReleaseEventSink = (flaky: FlakyReleaseWrites) => }), ).pipe(Layer.provide(layerTestEventSink)); +// Once armed, holds the next attach write until the writer is interrupted. +const layerPausingAttachEventSink = (pause: { + readonly armed: Ref.Ref; + readonly paused: Deferred.Deferred; +}) => + Layer.effect( + EventSink.EventSinkV2, + Effect.gen(function* () { + const delegate = yield* EventSink.EventSinkV2; + return EventSink.EventSinkV2.of({ + ...delegate, + write: (input) => + Effect.gen(function* () { + const attach = input.events.some((event) => event.type === "provider-session.attached"); + if (attach && (yield* Ref.getAndSet(pause.armed, false))) { + yield* Deferred.succeed(pause.paused, undefined); + return yield* Effect.never; + } + return yield* delegate.write(input); + }), + }); + }), + ).pipe(Layer.provide(layerTestEventSink)); + const CodexCapabilities: OrchestrationV2ProviderCapabilities = CodexProviderCapabilitiesV2; const ExclusiveCapabilities: OrchestrationV2ProviderCapabilities = { ...CodexCapabilities, @@ -244,11 +268,13 @@ function makeProviderThread(input: { readonly threadId: ThreadId; readonly providerSessionId: ProviderSessionId; readonly now: DateTime.Utc; + readonly nativeThreadId?: string; }): OrchestrationV2ProviderThread { + const nativeThreadId = input.nativeThreadId ?? "native-thread"; return { id: input.idAllocator.derive.providerThread({ driver: CODEX_DRIVER, - nativeThreadId: "native-thread", + nativeThreadId, }), driver: CODEX_DRIVER, providerInstanceId: modelSelection.instanceId, @@ -257,7 +283,7 @@ function makeProviderThread(input: { ownerNodeId: null, nativeThreadRef: { driver: CODEX_DRIVER, - nativeId: "native-thread", + nativeId: nativeThreadId, strength: "strong", }, nativeConversationHeadRef: null, @@ -294,9 +320,29 @@ function makeProviderAdapter( }) => Effect.Effect; readonly hasPendingBackgroundWork?: Effect.Effect; readonly hangSessionScopeClose?: boolean; + readonly startTurn?: Effect.Effect; readonly beforeUnload?: Effect.Effect; + /** Registers the process's closeCount finalizer before `beforeOpen` runs. */ + readonly spawnBeforeOpen?: boolean; + /** Completed when a hanging scope close reaches its wedged finalizer. */ + readonly scopeCloseReached?: Deferred.Deferred; } = {}, ): ProviderAdapterV2Shape { + const countClose = Effect.addFinalizer(() => + Ref.update(state, (current) => ({ + ...current, + closeCount: current.closeCount + 1, + })), + ); + // Registered after countClose so it runs first on scope close, wedging the + // close before the closeCount finalizer, like a provider process that never + // yields its message stream. + const hangClose = Effect.addFinalizer(() => + (options.scopeCloseReached === undefined + ? Effect.void + : Deferred.succeed(options.scopeCloseReached, undefined) + ).pipe(Effect.andThen(Effect.never)), + ); return { instanceId: ProviderInstanceId.make("codex"), driver: CODEX_DRIVER, @@ -304,10 +350,20 @@ function makeProviderAdapter( planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), openSession: (input) => Effect.gen(function* () { + if (options.spawnBeforeOpen === true) { + yield* countClose; + if (options.hangSessionScopeClose === true) yield* hangClose; + } + if (options.mcpConfigs !== undefined && options.spawnBeforeOpen === true) { + yield* Ref.update(options.mcpConfigs, (configs) => [ + ...configs, + McpProviderSession.readMcpProviderSession(input.threadId), + ]); + } if (options.beforeOpen !== undefined) { yield* options.beforeOpen(input); } - if (options.mcpConfigs !== undefined) { + if (options.mcpConfigs !== undefined && options.spawnBeforeOpen !== true) { yield* Ref.update(options.mcpConfigs, (configs) => [ ...configs, McpProviderSession.readMcpProviderSession(input.threadId), @@ -329,17 +385,9 @@ function makeProviderAdapter( eventQueues, }; }); - yield* Effect.addFinalizer(() => - Ref.update(state, (current) => ({ - ...current, - closeCount: current.closeCount + 1, - })), - ); - if (options.hangSessionScopeClose === true) { - // Registered last so it runs first on scope close, wedging the - // close before the closeCount finalizer, like a provider process - // that never yields its message stream. - yield* Effect.addFinalizer(() => Effect.never); + if (options.spawnBeforeOpen !== true) { + yield* countClose; + if (options.hangSessionScopeClose === true) yield* hangClose; } return { @@ -365,7 +413,7 @@ function makeProviderAdapter( ...current, resumeCount: current.resumeCount + 1, })).pipe(Effect.as(threadInput.providerThread)), - startTurn: () => Effect.void, + startTurn: () => options.startTurn ?? Effect.void, steerTurn: () => Effect.void, interruptTurn: () => Ref.update(state, (current) => ({ @@ -408,18 +456,29 @@ function layerTest(input: { }) => Effect.Effect; readonly failReleaseEventWrites?: boolean; readonly flakyReleaseWrites?: FlakyReleaseWrites; + readonly pauseAttachWrite?: Parameters[0]; + /** Once armed, holds the next credential lookup until it is interrupted. */ + readonly pauseResolve?: { + readonly armed: Ref.Ref; + readonly paused: Deferred.Deferred; + }; readonly hasPendingBackgroundWork?: Effect.Effect; readonly hangSessionScopeClose?: boolean; + readonly startTurn?: Effect.Effect; readonly beforeUnload?: Effect.Effect; + readonly spawnBeforeOpen?: boolean; + readonly scopeCloseReached?: Deferred.Deferred; readonly serverSettingsLayer?: ReturnType; readonly projectServiceLayer?: Layer.Layer; }) { const layerConfiguredEventSink = input.flakyReleaseWrites !== undefined ? layerFlakyReleaseEventSink(input.flakyReleaseWrites) - : input.failReleaseEventWrites - ? layerFailingReleaseEventSink - : layerTestEventSink; + : input.pauseAttachWrite !== undefined + ? layerPausingAttachEventSink(input.pauseAttachWrite) + : input.failReleaseEventWrites + ? layerFailingReleaseEventSink + : layerTestEventSink; const layerRegistry = ProviderAdapterRegistry.layerSingle( makeProviderAdapter(input.state, { failEventStream: input.failEventStream ?? false, @@ -432,9 +491,18 @@ function layerTest(input: { ...(input.hangSessionScopeClose === undefined ? {} : { hangSessionScopeClose: input.hangSessionScopeClose }), + ...(input.startTurn === undefined ? {} : { startTurn: input.startTurn }), ...(input.beforeUnload === undefined ? {} : { beforeUnload: input.beforeUnload }), + ...(input.spawnBeforeOpen === undefined ? {} : { spawnBeforeOpen: input.spawnBeforeOpen }), + ...(input.scopeCloseReached === undefined + ? {} + : { scopeCloseReached: input.scopeCloseReached }), }), ); + const layerConfiguredMcpRegistry = + input.pauseResolve === undefined + ? layerTestMcpRegistry + : layerPausingMcpRegistry(input.pauseResolve); const layerProviderEventIngestorTest = ProviderEventIngestor.layer.pipe( Layer.provide( Layer.mergeAll( @@ -449,7 +517,7 @@ function layerTest(input: { layerTestStores, layerConfiguredEventSink, IdAllocator.layer, - layerTestMcpRegistry, + layerConfiguredMcpRegistry, ProviderSessionManager.layerWithOptions({ idleTimeoutMs: input.idleTimeoutMs, ...(input.maxIdlePinMs === undefined ? {} : { maxIdlePinMs: input.maxIdlePinMs }), @@ -460,7 +528,7 @@ function layerTest(input: { layerConfiguredEventSink, IdAllocator.layer, layerProviderEventIngestorTest, - layerTestMcpRegistry, + layerConfiguredMcpRegistry, layerTestStores, ...(input.serverSettingsLayer === undefined ? [] : [input.serverSettingsLayer]), ...(input.projectServiceLayer === undefined ? [] : [input.projectServiceLayer]), @@ -489,6 +557,28 @@ const layerTestMcpRegistry = Layer.effect( Layer.provide(NodeServices.layer), ); +const layerPausingMcpRegistry = (pause: { + readonly armed: Ref.Ref; + readonly paused: Deferred.Deferred; +}) => + Layer.effect( + McpSessionRegistry.McpSessionRegistry, + Effect.gen(function* () { + const delegate = yield* McpSessionRegistry.McpSessionRegistry; + return McpSessionRegistry.McpSessionRegistry.of({ + ...delegate, + resolve: (rawToken) => + Ref.getAndSet(pause.armed, false).pipe( + Effect.flatMap((armed) => + armed + ? Deferred.succeed(pause.paused, undefined).pipe(Effect.andThen(Effect.never)) + : delegate.resolve(rawToken), + ), + ), + }); + }), + ).pipe(Layer.provide(layerTestMcpRegistry)); + function makeBrowserAccessProject(projectId: ProjectId): Project { return { id: projectId, @@ -997,6 +1087,498 @@ it.effect("ProviderSessionManagerV2 opens a duplicate session only once", () => }), ); +it.effect("ProviderSessionManagerV2 cleans up an open interrupted mid-handshake", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const mcpConfigs = yield* Ref.make< + ReadonlyArray + >([]); + const handshakeStarted = yield* Deferred.make(); + const holdHandshake = yield* Ref.make(true); + const effect = Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const registry = yield* McpSessionRegistry.McpSessionRegistry; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread-provider-session-manager-interrupted-open"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + + const opening = yield* manager + .open({ threadId, providerSessionId, modelSelection, runtimePolicy }) + .pipe(Effect.forkScoped); + yield* Deferred.await(handshakeStarted); + const issued = (yield* Ref.get(mcpConfigs)).at(-1); + const token = issued?.authorizationHeader.replace(/^Bearer\s+/, ""); + assert.isDefined(token); + assert.isDefined(yield* registry.resolve(token!)); + + // A Stop while the provider is still starting. + yield* Fiber.interrupt(opening); + + // The process started for this open is stopped, and the credential minted + // for it revoked. + assert.equal((yield* Ref.get(state)).closeCount, 1); + assert.isUndefined(yield* registry.resolve(token!)); + assert.isUndefined(McpProviderSession.readMcpProviderSession(threadId)); + assert.isTrue(Option.isNone(yield* manager.get(providerSessionId))); + + // Nothing of the interrupted open is left behind: the next open starts a + // fresh process with a fresh credential that a later release revokes, + // which a leaked reservation would prevent. + yield* Ref.set(holdHandshake, false); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + const replacement = (yield* Ref.get(mcpConfigs)).at(-1); + const replacementToken = replacement?.authorizationHeader.replace(/^Bearer\s+/, ""); + assert.isDefined(replacementToken); + assert.notEqual(replacementToken, token); + const projection = yield* projectionStore.getThreadProjection(threadId); + assert.equal(projection.providerSessions.at(-1)?.status, "ready"); + + yield* manager.close(providerSessionId); + assert.isUndefined(yield* registry.resolve(replacementToken!)); + }); + + yield* effect.pipe( + Effect.provide( + layerTest({ + state, + idleTimeoutMs: 60_000, + mcpConfigs, + beforeOpen: () => + Ref.get(holdHandshake).pipe( + Effect.flatMap((hold) => + hold + ? Deferred.succeed(handshakeStarted, undefined).pipe(Effect.andThen(Effect.never)) + : Effect.void, + ), + ), + // The process is spawned before the handshake that is interrupted. + spawnBeforeOpen: true, + }), + ), + ); + }), +); + +it.effect("ProviderSessionManagerV2 cleans up an interrupted open whose scope close hangs", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const mcpConfigs = yield* Ref.make< + ReadonlyArray + >([]); + const handshakeStarted = yield* Deferred.make(); + const scopeCloseReached = yield* Deferred.make(); + const releaseRetry = yield* Deferred.make(); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const registry = yield* McpSessionRegistry.McpSessionRegistry; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread-provider-session-manager-interrupted-hung-open"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + + const opening = yield* manager + .open({ threadId, providerSessionId, modelSelection, runtimePolicy }) + .pipe(Effect.forkChild); + yield* Deferred.await(handshakeStarted); + const issued = (yield* Ref.get(mcpConfigs)).at(-1); + const token = issued?.authorizationHeader.replace(/^Bearer\s+/, ""); + assert.isDefined(token); + + // The interrupt starts cleanup; the scope close then never finishes. + const interrupter = yield* Fiber.interrupt(opening).pipe(Effect.forkChild); + yield* Deferred.await(scopeCloseReached); + + // The session cleanup already ran, ahead of the stuck close. + assert.isUndefined(yield* registry.resolve(token!)); + assert.isUndefined(McpProviderSession.readMcpProviderSession(threadId)); + assert.equal((yield* Ref.get(state)).closeCount, 0); + + // The close is time-boxed, so the interrupter returns and the session's + // open lock is free for the next open. + yield* TestClock.adjust("30 seconds"); + yield* Fiber.join(interrupter); + yield* Deferred.succeed(releaseRetry, undefined); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + assert.equal((yield* Ref.get(state)).openCount, 1); + + // Its close hangs as well; release it while the test clock can still move. + const stopping = yield* manager.shutdown.pipe(Effect.forkChild); + yield* TestClock.adjust("30 seconds"); + yield* Fiber.join(stopping); + }).pipe( + Effect.provide( + layerTest({ + state, + idleTimeoutMs: 60_000, + mcpConfigs, + // The first open hangs in its handshake; the retry completes. + beforeOpen: () => + Deferred.isDone(handshakeStarted).pipe( + Effect.flatMap((retry) => + retry + ? Deferred.await(releaseRetry) + : Deferred.succeed(handshakeStarted, undefined).pipe( + Effect.andThen(Effect.never), + ), + ), + ), + spawnBeforeOpen: true, + hangSessionScopeClose: true, + scopeCloseReached, + }), + ), + ); + }), +); + +it.effect("ProviderSessionManagerV2 releases an idle session whose turn start was stopped", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const startTurnReached = yield* Deferred.make(); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread-provider-session-manager-stopped-turn-start"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + const runtime = yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + const runId = idAllocator.derive.run({ threadId, ordinal: 1 }); + const starting = yield* runtime + .startTurn({ + appThread: (yield* projectionStore.getThreadProjection(threadId)).thread, + threadId, + runId, + runOrdinal: 1, + providerTurnOrdinal: 1, + attemptId: idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }), + rootNodeId: idAllocator.derive.rootNode({ runId }), + providerThread: makeProviderThread({ idAllocator, threadId, providerSessionId, now }), + message: { + createdBy: "user", + creationSource: "web", + messageId: yield* idAllocator.allocate.message({ threadId, ordinal: 1 }), + text: "stopped before the provider accepted it", + attachments: [], + }, + modelSelection, + runtimePolicy, + }) + .pipe(Effect.forkChild); + yield* Deferred.await(startTurnReached); + + // Stop lands while the provider is still accepting the turn. No terminal + // event follows, so the session must count itself idle again. + yield* Fiber.interrupt(starting); + yield* TestClock.adjust("2 seconds"); + yield* Effect.yieldNow; + assert.isTrue(Option.isNone(yield* manager.get(providerSessionId))); + assert.equal((yield* Ref.get(state)).closeCount, 1); + }).pipe( + Effect.provide( + layerTest({ + state, + idleTimeoutMs: 1000, + startTurn: Deferred.succeed(startTurnReached, undefined).pipe( + Effect.andThen(Effect.never), + ), + }), + ), + ); + }), +); + +it.effect( + "ProviderSessionManagerV2 keeps a shared session busy when another thread's start is stopped early", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const projectId = ProjectId.make("project-provider-session-manager-shared-stopped-start"); + // Blocks the next project read, so a re-attach can be stopped before its + // start marks the session busy. + const holdProjectRead = yield* Ref.make(false); + const projectReadHeld = yield* Deferred.make(); + const projectServiceLayer = Layer.mock(ProjectService.ProjectService)({ + getById: (requestedProjectId) => + Ref.get(holdProjectRead).pipe( + Effect.flatMap((hold) => + hold + ? Deferred.succeed(projectReadHeld, undefined).pipe(Effect.andThen(Effect.never)) + : Effect.succeed(Option.some(makeBrowserAccessProject(requestedProjectId))), + ), + ), + }); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadA = ThreadId.make("thread-provider-session-manager-shared-stopped-a"); + const threadB = ThreadId.make("thread-provider-session-manager-shared-stopped-b"); + const providerSessionId = idAllocator.derive.providerSession({ + providerInstanceId: modelSelection.instanceId, + }); + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator, threadId: threadA, now, projectId }), + yield* makeThreadCreatedEvent({ idAllocator, threadId: threadB, now, projectId }), + ], + }); + const runtime = yield* manager.open({ + threadId: threadA, + providerSessionId, + modelSelection, + runtimePolicy, + }); + yield* manager.open({ + threadId: threadB, + providerSessionId, + modelSelection, + runtimePolicy, + }); + const startTurn = (threadId: ThreadId) => + Effect.gen(function* () { + const runId = idAllocator.derive.run({ threadId, ordinal: 1 }); + return yield* runtime.startTurn({ + appThread: (yield* projectionStore.getThreadProjection(threadId)).thread, + threadId, + runId, + runOrdinal: 1, + providerTurnOrdinal: 1, + attemptId: idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }), + rootNodeId: idAllocator.derive.rootNode({ runId }), + providerThread: makeProviderThread({ idAllocator, threadId, providerSessionId, now }), + message: { + createdBy: "user", + creationSource: "web", + messageId: yield* idAllocator.allocate.message({ threadId, ordinal: 1 }), + text: "turn", + attachments: [], + }, + modelSelection, + runtimePolicy, + }); + }); + + // B's turn is accepted and still running. + yield* startTurn(threadB); + + // A detaches, and its next start is stopped while re-attaching, before + // it marks the session busy. + yield* manager.detach({ providerSessionId, threadId: threadA }); + yield* Ref.set(holdProjectRead, true); + const startingA = yield* startTurn(threadA).pipe(Effect.forkChild); + yield* Deferred.await(projectReadHeld); + yield* Fiber.interrupt(startingA); + + // B's running turn keeps the session busy past the idle timeout. + yield* TestClock.adjust("2 seconds"); + yield* Effect.yieldNow; + assert.isTrue(Option.isSome(yield* manager.get(providerSessionId))); + assert.equal((yield* Ref.get(state)).closeCount, 0); + }).pipe( + Effect.provide( + layerTest({ + state, + idleTimeoutMs: 1000, + projectServiceLayer, + serverSettingsLayer: ServerSettings.layerTest({ + projectSettingsOverrides: { [projectId]: { enableAgentBrowserAccess: true } }, + }), + }), + ), + ); + }), +); + +it.effect( + "ProviderSessionManagerV2 keeps a shared session busy when a stopped start still ends its turn", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + // A's adapter start installs its turn, then is stopped mid-request; like + // OpenCode2, the adapter still ends that turn with turn.terminal later. + const holdStart = yield* Ref.make(false); + const startHeld = yield* Deferred.make(); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadA = ThreadId.make("thread-provider-session-manager-stopped-ended-a"); + const threadB = ThreadId.make("thread-provider-session-manager-stopped-ended-b"); + const providerSessionId = idAllocator.derive.providerSession({ + providerInstanceId: modelSelection.instanceId, + }); + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator, threadId: threadA, now }), + yield* makeThreadCreatedEvent({ idAllocator, threadId: threadB, now }), + ], + }); + const runtime = yield* manager.open({ + threadId: threadA, + providerSessionId, + modelSelection, + runtimePolicy, + }); + yield* manager.open({ + threadId: threadB, + providerSessionId, + modelSelection, + runtimePolicy, + }); + const providerThreadOf = (threadId: ThreadId) => + makeProviderThread({ + idAllocator, + threadId, + providerSessionId, + now, + nativeThreadId: `native-${threadId}`, + }); + const startTurn = (threadId: ThreadId) => + Effect.gen(function* () { + const runId = idAllocator.derive.run({ threadId, ordinal: 1 }); + return yield* runtime.startTurn({ + appThread: (yield* projectionStore.getThreadProjection(threadId)).thread, + threadId, + runId, + runOrdinal: 1, + providerTurnOrdinal: 1, + attemptId: idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }), + rootNodeId: idAllocator.derive.rootNode({ runId }), + providerThread: providerThreadOf(threadId), + message: { + createdBy: "user", + creationSource: "web", + messageId: yield* idAllocator.allocate.message({ threadId, ordinal: 1 }), + text: "turn", + attachments: [], + }, + modelSelection, + runtimePolicy, + }); + }); + + // B's turn is accepted and still running. + yield* startTurn(threadB); + + // A's start is stopped after the adapter began the turn. + yield* Ref.set(holdStart, true); + const startingA = yield* startTurn(threadA).pipe(Effect.forkChild); + yield* Deferred.await(startHeld); + yield* Fiber.interrupt(startingA); + + // The adapter ends A's turn anyway. + const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); + assert.isDefined(queue); + yield* Queue.offer(queue!, { + type: "turn.terminal", + driver: CODEX_DRIVER, + providerThreadId: providerThreadOf(threadA).id, + providerTurnId: idAllocator.derive.providerTurn({ + driver: CODEX_DRIVER, + nativeTurnId: "native-turn-stopped-a", + }), + runOrdinal: 1, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + + // B's running turn keeps the session busy past the idle timeout. + yield* TestClock.adjust("2 seconds"); + yield* Effect.yieldNow; + assert.isTrue(Option.isSome(yield* manager.get(providerSessionId))); + assert.equal((yield* Ref.get(state)).closeCount, 0); + }).pipe( + Effect.provide( + layerTest({ + state, + idleTimeoutMs: 1000, + startTurn: Ref.get(holdStart).pipe( + Effect.flatMap((hold) => + hold + ? Deferred.succeed(startHeld, undefined).pipe(Effect.andThen(Effect.never)) + : Effect.void, + ), + ), + }), + ), + ); + }), +); + +it.effect("ProviderSessionManagerV2 stops a session still opening when its layer shuts down", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const handshakeStarted = yield* Deferred.make(); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread-provider-session-manager-shutdown-during-open"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + // Detached, like an open the layer does not own: only the session + // scope's parent can stop its process when the layer closes. + yield* manager + .open({ threadId, providerSessionId, modelSelection, runtimePolicy }) + .pipe(Effect.forkDetach); + yield* Deferred.await(handshakeStarted); + assert.equal((yield* Ref.get(state)).closeCount, 0); + }).pipe( + Effect.provide( + layerTest({ + state, + idleTimeoutMs: 60_000, + beforeOpen: () => + Deferred.succeed(handshakeStarted, undefined).pipe(Effect.andThen(Effect.never)), + spawnBeforeOpen: true, + }), + ), + ); + + assert.equal((yield* Ref.get(state)).closeCount, 1); + }), +); + it.effect("ProviderSessionManagerV2 releases live sessions when its layer shuts down", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); @@ -1368,6 +1950,228 @@ it.effect("ProviderSessionManagerV2 revokes MCP credentials when release persist }), ); +it.effect( + "ProviderSessionManagerV2 keeps a thread attached when an earlier attach of it is interrupted", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const armed = yield* Ref.make(false); + const paused = yield* Deferred.make(); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const registry = yield* McpSessionRegistry.McpSessionRegistry; + const now = yield* DateTime.now; + const owner = ThreadId.make("thread-provider-session-manager-attach-race-owner"); + const threadId = ThreadId.make("thread-provider-session-manager-attach-race"); + const providerSessionId = idAllocator.derive.providerSession({ + providerInstanceId: modelSelection.instanceId, + }); + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator, threadId: owner, now }), + yield* makeThreadCreatedEvent({ idAllocator, threadId, now }), + ], + }); + const runtime = yield* manager.open({ + threadId: owner, + providerSessionId, + modelSelection, + runtimePolicy, + }); + const resume = runtime.resumeThread({ + threadId, + providerThread: makeProviderThread({ + idAllocator, + threadId, + providerSessionId, + now, + nativeThreadId: "native-attach-race", + }), + }); + + // The first attach of the thread is stopped after attaching it, while + // a second attach of the same thread is already on its way. + yield* Ref.set(armed, true); + const first = yield* resume.pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(paused); + const second = yield* resume.pipe(Effect.forkChild({ startImmediately: true })); + yield* Fiber.interrupt(first); + yield* Fiber.join(second); + + // The second attach owns the thread's attachment and credential. + const config = McpProviderSession.readMcpProviderSession(threadId); + assert.isDefined(config); + const token = config!.authorizationHeader.replace(/^Bearer\s+/, ""); + assert.equal((yield* registry.resolve(token))?.thread.threadId, threadId); + }).pipe( + Effect.provide( + layerTest({ state, idleTimeoutMs: 60_000, pauseAttachWrite: { armed, paused } }), + ), + ); + }), +); + +it.effect( + "ProviderSessionManagerV2 keeps a replacement session's attachment when a stale attach is interrupted", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const armed = yield* Ref.make(false); + const paused = yield* Deferred.make(); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const registry = yield* McpSessionRegistry.McpSessionRegistry; + const eventStore = yield* EventStore.EventStoreV2; + const now = yield* DateTime.now; + const owner = ThreadId.make("thread-provider-session-manager-stale-attach-owner"); + const threadId = ThreadId.make("thread-provider-session-manager-stale-attach"); + const providerSessionId = idAllocator.derive.providerSession({ + providerInstanceId: modelSelection.instanceId, + }); + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator, threadId: owner, now }), + yield* makeThreadCreatedEvent({ idAllocator, threadId, now }), + ], + }); + const runtime = yield* manager.open({ + threadId: owner, + providerSessionId, + modelSelection, + runtimePolicy, + }); + + // An attach of the thread to this session stalls mid-write... + yield* Ref.set(armed, true); + const stale = yield* runtime + .resumeThread({ + threadId, + providerThread: makeProviderThread({ + idAllocator, + threadId, + providerSessionId, + now, + nativeThreadId: "native-stale-attach", + }), + }) + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(paused); + + // ...while the session is replaced and the thread opens the new one. + yield* manager.release({ providerSessionId, reason: "runtime_error" }); + const replacement = yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + const config = McpProviderSession.readMcpProviderSession(threadId); + assert.isDefined(config); + + yield* Fiber.interrupt(stale); + + // The replacement keeps the thread and its credential. + assert.equal( + McpProviderSession.readMcpProviderSession(threadId)?.providerSessionId, + config!.providerSessionId, + ); + const token = config!.authorizationHeader.replace(/^Bearer\s+/, ""); + assert.equal((yield* registry.resolve(token))?.thread.threadId, threadId); + // Still attached: resuming on the replacement does not attach the thread again. + const attachedEvents = eventStore + .read({ threadId, eventType: "provider-session.attached" }) + .pipe( + Stream.runCollect, + Effect.map((events) => events.length), + ); + const attachedBefore = yield* attachedEvents; + yield* replacement.resumeThread({ + threadId, + providerThread: makeProviderThread({ + idAllocator, + threadId, + providerSessionId, + now, + nativeThreadId: "native-stale-attach", + }), + }); + assert.equal(yield* attachedEvents, attachedBefore); + }).pipe( + Effect.provide( + layerTest({ state, idleTimeoutMs: 60_000, pauseAttachWrite: { armed, paused } }), + ), + ); + }), +); + +it.effect( + "ProviderSessionManagerV2 revokes a reused credential after a resume stopped while checking it", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const armed = yield* Ref.make(false); + const paused = yield* Deferred.make(); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const registry = yield* McpSessionRegistry.McpSessionRegistry; + const now = yield* DateTime.now; + const owner = ThreadId.make("thread-provider-session-manager-resolve-stop-owner"); + const threadId = ThreadId.make("thread-provider-session-manager-resolve-stop"); + const providerSessionId = idAllocator.derive.providerSession({ + providerInstanceId: modelSelection.instanceId, + }); + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator, threadId: owner, now }), + yield* makeThreadCreatedEvent({ idAllocator, threadId, now }), + ], + }); + const runtime = yield* manager.open({ + threadId: owner, + providerSessionId, + modelSelection, + runtimePolicy, + }); + const resume = runtime.resumeThread({ + threadId, + providerThread: makeProviderThread({ + idAllocator, + threadId, + providerSessionId, + now, + nativeThreadId: "native-resolve-stop", + }), + }); + // The thread gets a credential, then detaches and keeps it for a re-attach. + yield* resume; + yield* manager.detach({ providerSessionId, threadId }); + const config = McpProviderSession.readMcpProviderSession(threadId); + assert.isDefined(config); + const token = config!.authorizationHeader.replace(/^Bearer\s+/, ""); + + // A re-attach is stopped while it checks whether that credential is reusable. + yield* Ref.set(armed, true); + const stopped = yield* resume.pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(paused); + yield* Fiber.interrupt(stopped); + + // Nothing holds the credential now, so a terminal release revokes it. + yield* manager.release({ providerSessionId, reason: "manual_shutdown" }); + assert.isUndefined(yield* registry.resolve(token)); + }).pipe( + Effect.provide( + layerTest({ state, idleTimeoutMs: 60_000, pauseResolve: { armed, paused } }), + ), + ); + }), +); + it.effect("ProviderSessionManagerV2 duplicate detach preserves replacement MCP credentials", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); @@ -3011,17 +3815,20 @@ it.effect( providerInstanceId: modelSelection.instanceId, threadId: firstThreadId, }); + // Each thread has its own native thread on the shared session. const firstProviderThread = makeProviderThread({ idAllocator, threadId: firstThreadId, providerSessionId, now, + nativeThreadId: "native-thread-a", }); const secondProviderThread = makeProviderThread({ idAllocator, threadId: secondThreadId, providerSessionId, now, + nativeThreadId: "native-thread-b", }); const firstRunId = idAllocator.derive.run({ threadId: firstThreadId, ordinal: 1 }); const secondRunId = idAllocator.derive.run({ threadId: secondThreadId, ordinal: 1 }); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 5730432aad87ed5015c05c194004593f6985be0a..f84441129cc3ae10ea738bd3ebce245ea877ec1d 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -9,6 +9,7 @@ import { ProviderInstanceId, ProviderSessionId, ThreadId, + type ProviderThreadId, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Clock from "effect/Clock"; @@ -59,6 +60,10 @@ import * as ProjectionStore from "./ProjectionStore.ts"; const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1000; const DEFAULT_MAX_IDLE_PIN_MS = 4 * 60 * 60 * 1000; const RELEASE_SCOPE_CLOSE_TIMEOUT_MS = 30 * 1000; + +/** The identity a turn's start and its `turn.terminal` share. */ +const busyTurnKey = (providerThreadId: ProviderThreadId, runOrdinal: number) => + `${providerThreadId}#${runOrdinal}`; const UNLOAD_THREAD_TIMEOUT_MS = 10 * 1000; export const ProviderSessionReleaseReason = Schema.Literals([ @@ -213,7 +218,12 @@ interface LiveSessionEntry { readonly requestEventPermit: Semaphore.Semaphore; readonly scope: Scope.Closeable; readonly idleGeneration: number; - readonly busyCount: number; + /** + * Turns this session is running, keyed by `busyTurnKey`. A turn's start adds + * it and its `turn.terminal` (or a failed start) removes it, so a turn can + * only clear itself and the session is idle when the set is empty. + */ + readonly busyTurns: ReadonlySet; readonly lastActivityAtMs: number; readonly idleFiber: Fiber.Fiber | null; /** Set when idle release is deferred for pending background work; bounds total deferral. */ @@ -472,7 +482,17 @@ export const layerWithOptions = ( // revoke the credential between validation and reservation. reserveMcpCredential(threadId, existing.providerSessionId); const rawToken = existing.authorizationHeader.replace(/^Bearer\s+/, ""); - const resolved = yield* mcpSessionRegistry.resolve(rawToken); + // The caller only learns of the reservation once this returns, + // so a stop while resolving must drop it here. + const resolved = yield* mcpSessionRegistry + .resolve(rawToken) + .pipe( + Effect.onInterrupt(() => + Effect.sync(() => + dropMcpCredentialReservation(threadId, existing.providerSessionId), + ), + ), + ); if ( resolved !== undefined && resolved.thread.threadId === threadId && @@ -842,7 +862,8 @@ export const layerWithOptions = ( } if ( input.onlyIfIdleGeneration !== undefined && - (existing.busyCount > 0 || existing.idleGeneration !== input.onlyIfIdleGeneration) + (existing.busyTurns.size > 0 || + existing.idleGeneration !== input.onlyIfIdleGeneration) ) { return ["kept", current] as const; } @@ -861,6 +882,45 @@ export const layerWithOptions = ( ] as const; }); + // Scope close can wedge on a misbehaving adapter finalizer (e.g. a + // provider process that never yields its message stream). Time-box it so + // the caller, and any lock or worker it holds, moves on and leaves a + // diagnosable trail. A close that finishes late is still logged. + const closeScopeWithin = ( + scope: Scope.Closeable, + annotations: { readonly providerSessionId?: ProviderSessionId; readonly reason: string }, + ) => + Effect.gen(function* () { + const closeFiber = yield* Scope.close(scope, Exit.void).pipe( + Effect.exit, + Effect.forkDetach({ startImmediately: true }), + ); + const closeExit = yield* Fiber.join(closeFiber).pipe( + Effect.timeoutOption(RELEASE_SCOPE_CLOSE_TIMEOUT_MS), + ); + if (Option.isNone(closeExit)) { + yield* Effect.logWarning("orchestration-v2.provider-session-scope-close-timeout", { + ...annotations, + timeoutMs: RELEASE_SCOPE_CLOSE_TIMEOUT_MS, + }); + yield* Fiber.join(closeFiber).pipe( + Effect.flatMap((exit) => + Exit.isFailure(exit) + ? Effect.logWarning("orchestration-v2.provider-session-scope-close-failed", { + ...annotations, + cause: exit.cause, + }) + : Effect.logInfo( + "orchestration-v2.provider-session-scope-close-completed-late", + annotations, + ), + ), + Effect.forkDetach, + ); + } + return closeExit; + }); + const releaseEntry = (input: { readonly providerSessionId: ProviderSessionId; readonly reason: ProviderSessionReleaseReason; @@ -889,49 +949,10 @@ export const layerWithOptions = ( input.detail ?? `Provider session released: ${input.reason}.`, ); } - // Scope close can wedge on a misbehaving adapter finalizer - // (e.g. a provider process that never yields its message - // stream). Time-box it so release still persists released - // events and leaves a diagnosable trail instead of silently - // parking the session as "ready" forever. - const closeFiber = yield* Scope.close(entry.scope, Exit.void).pipe( - Effect.exit, - Effect.forkDetach({ startImmediately: true }), - ); - const closeExit = yield* Fiber.join(closeFiber).pipe( - Effect.timeoutOption(RELEASE_SCOPE_CLOSE_TIMEOUT_MS), - ); - if (Option.isNone(closeExit)) { - yield* Effect.logWarning( - "orchestration-v2.provider-session-scope-close-timeout", - { - providerSessionId: input.providerSessionId, - reason: input.reason, - timeoutMs: RELEASE_SCOPE_CLOSE_TIMEOUT_MS, - }, - ); - yield* Fiber.join(closeFiber).pipe( - Effect.flatMap((exit) => - Exit.isFailure(exit) - ? Effect.logWarning( - "orchestration-v2.provider-session-scope-close-failed", - { - providerSessionId: input.providerSessionId, - reason: input.reason, - cause: exit.cause, - }, - ) - : Effect.logInfo( - "orchestration-v2.provider-session-scope-close-completed-late", - { - providerSessionId: input.providerSessionId, - reason: input.reason, - }, - ), - ), - Effect.forkDetach, - ); - } + const closeExit = yield* closeScopeWithin(entry.scope, { + providerSessionId: input.providerSessionId, + reason: input.reason, + }); const records = { entry, reason: input.reason, @@ -1022,7 +1043,7 @@ export const layerWithOptions = ( const entry = current.get(key); if ( entry === undefined || - entry.busyCount > 0 || + entry.busyTurns.size > 0 || entry.idleGeneration !== input.generation ) { return; @@ -1044,7 +1065,7 @@ export const layerWithOptions = ( const latestEntry = latest.get(key); if ( latestEntry === undefined || - latestEntry.busyCount > 0 || + latestEntry.busyTurns.size > 0 || latestEntry.idleGeneration !== input.generation || latestEntry.runtime !== probedRuntime ) { @@ -1076,7 +1097,7 @@ export const layerWithOptions = ( } // hasPendingBackgroundWork yields to the adapter, so the idle // decision above can go stale; the generation guard revalidates - // busyCount and idleGeneration inside releaseEntry's atomic + // busyTurns and idleGeneration inside releaseEntry's atomic // entry removal. yield* releaseEntry({ providerSessionId: input.providerSessionId, @@ -1113,7 +1134,7 @@ export const layerWithOptions = ( const key = sessionKey(providerSessionId); const current = yield* Ref.get(sessions); const entry = current.get(key); - if (entry === undefined || entry.busyCount > 0) { + if (entry === undefined || entry.busyTurns.size > 0) { return; } @@ -1126,7 +1147,7 @@ export const layerWithOptions = ( const lastActivityAtMs = yield* Clock.currentTimeMillis; yield* Ref.update(sessions, (latest) => { const latestEntry = latest.get(key); - if (latestEntry === undefined || latestEntry.busyCount > 0) { + if (latestEntry === undefined || latestEntry.busyTurns.size > 0) { return latest; } const updated = new Map(latest); @@ -1164,6 +1185,7 @@ export const layerWithOptions = ( }), ); + /** Returns the runtime the thread was attached to, or undefined if it already was. */ const attachThread = (input: { readonly providerSessionId: ProviderSessionId; readonly threadId: ThreadId; @@ -1173,25 +1195,34 @@ export const layerWithOptions = ( Ref.modify(sessions, (current) => { const entry = current.get(sessionKey(input.providerSessionId)); if (entry === undefined || entry.attachedThreadIds.has(input.threadId)) { - return [false, current] as const; + return [undefined, current] as const; } const updated = new Map(current); updated.set(sessionKey(input.providerSessionId), { ...entry, attachedThreadIds: new Set([...entry.attachedThreadIds, input.threadId]), }); - return [true, updated] as const; + return [entry.runtime, updated] as const; }), ); + /** + * Undoes an attach to `runtime`. A replacement session that reopened under + * the same id since is left alone. + */ const removeThreadAttachment = (input: { readonly providerSessionId: ProviderSessionId; readonly threadId: ThreadId; + readonly runtime: ProviderAdapterV2SessionRuntime; }) => Ref.update(sessions, (current) => { const key = sessionKey(input.providerSessionId); const entry = current.get(key); - if (entry === undefined || !entry.attachedThreadIds.has(input.threadId)) { + if ( + entry === undefined || + entry.runtime !== input.runtime || + !entry.attachedThreadIds.has(input.threadId) + ) { return current; } const attachedThreadIds = new Set(entry.attachedThreadIds); @@ -1245,6 +1276,7 @@ export const layerWithOptions = ( readonly providerInstanceId: ProviderInstanceId; }) => Effect.suspend(() => { + let attachedTo: ProviderAdapterV2SessionRuntime | undefined; let preparedForCleanup: PreparedMcpCredential | undefined; let reservationDropped = false; const dropReservation = () => { @@ -1253,12 +1285,17 @@ export const layerWithOptions = ( dropMcpCredentialReservation(input.threadId, preparedForCleanup.mcpCredentialId); } }; - return Effect.gen(function* () { - const attached = yield* threadAttachment.withLock( - threadAttachmentKey(input), - attachThread(input), + // The whole attach, including undoing a failed one, holds the + // thread's lock: a concurrent attach of the same thread waits, so it + // never sees an attachment that this call is about to roll back. + const attach = Effect.gen(function* () { + const attached = yield* attachThread(input).pipe( + // Recorded with no gap for an interrupt: cleanup undoes only an + // attach this call made, never one an earlier open made. + Effect.tap((runtime) => Effect.sync(() => (attachedTo = runtime))), + Effect.uninterruptible, ); - if (attached) { + if (attached !== undefined) { const prepared = yield* prepareMcpSession(input.threadId, input.providerInstanceId); preparedForCleanup = prepared; if (prepared.mcpCredentialId !== undefined) { @@ -1288,22 +1325,46 @@ export const layerWithOptions = ( } } }).pipe( - Effect.tapError(() => - removeThreadAttachment(input).pipe( - // Revoke only a credential this attach freshly minted: a REUSED - // credential is by definition held by another live provider - // process, and revoking it thread-wide would break that - // process's MCP client mid-conversation. - Effect.andThen( - Effect.suspend(() => { - dropReservation(); - return preparedForCleanup?.issued === true - ? clearMcpSession(input.threadId, preparedForCleanup.mcpCredentialId) - : Effect.void; - }), - ), - ), + // An interrupted attach is undone too, so the next attach writes + // the attachment instead of finding the thread already attached. + Effect.onError(() => + attachedTo === undefined + ? Effect.void + : removeThreadAttachment({ ...input, runtime: attachedTo }).pipe( + Effect.andThen( + Effect.suspend(() => { + dropReservation(); + // Revoke only a credential this attach freshly minted: a + // REUSED credential is by definition held by another + // live provider process, and revoking it thread-wide + // would break that process's MCP client mid-conversation. + if (preparedForCleanup?.issued !== true) return Effect.void; + const mcpCredentialId = preparedForCleanup.mcpCredentialId; + const attachedRuntime = attachedTo; + // As in release: a replacement session (or an open + // configuring one) may have taken the credential up. + return Ref.get(sessions).pipe( + Effect.flatMap((current) => + (mcpCredentialId !== undefined && + isMcpCredentialReserved(input.threadId, mcpCredentialId)) || + Array.from(current.values()).some( + (other) => + other.runtime !== attachedRuntime && + (other.attachedThreadIds.has(input.threadId) || + (mcpCredentialId !== undefined && + other.mcpCredentialIdByThread.get(input.threadId) === + mcpCredentialId)), + ) + ? Effect.void + : clearMcpSession(input.threadId, mcpCredentialId), + ), + ); + }), + ), + ), ), + ); + return threadAttachment.withLock(threadAttachmentKey(input), attach).pipe( // The entry's own record (written above while the thread is // attached) guards the credential from here on; the reservation // is only needed until then. Ensuring covers defects/interrupts. @@ -1311,7 +1372,7 @@ export const layerWithOptions = ( ); }); - const markBusy = (providerSessionId: ProviderSessionId) => + const markBusy = (providerSessionId: ProviderSessionId, turnKey: string) => withActivityError( providerSessionId, Effect.gen(function* () { @@ -1325,7 +1386,7 @@ export const layerWithOptions = ( const updated = new Map(current); updated.set(key, { ...entry, - busyCount: entry.busyCount + 1, + busyTurns: new Set(entry.busyTurns).add(turnKey), idleFiber: null, lastActivityAtMs: now, pinnedSinceMs: null, @@ -1336,7 +1397,10 @@ export const layerWithOptions = ( }), ); - const markIdle = (providerSessionId: ProviderSessionId) => + // Clearing a turn that is not marked busy (one whose failed start already + // cleared it, or a subagent turn the manager never started) only + // records activity. + const markIdle = (providerSessionId: ProviderSessionId, turnKey: string) => withActivityError( providerSessionId, Effect.gen(function* () { @@ -1347,10 +1411,12 @@ export const layerWithOptions = ( if (entry === undefined) { return current; } + const busyTurns = new Set(entry.busyTurns); + busyTurns.delete(turnKey); const updated = new Map(current); updated.set(key, { ...entry, - busyCount: Math.max(0, entry.busyCount - 1), + busyTurns, lastActivityAtMs: now, }); return updated; @@ -1536,13 +1602,32 @@ export const layerWithOptions = ( providerInstanceId: runtime.instanceId, }), ).pipe( - Effect.andThen(observeActivity(providerSessionId, markBusy(providerSessionId))), + // A start that fails or is stopped may never emit turn.terminal, + // so it clears its own turn or the session never goes idle. If + // the adapter emits the terminal anyway, clearing the same turn + // again changes nothing, so another thread's turn on a shared + // session stays busy either way. Effect.andThen( - runtime.startTurn(input).pipe(turnMetrics("send", input.modelSelection.model)), - ), - Effect.catch((error) => - observeActivity(providerSessionId, markIdle(providerSessionId)).pipe( - Effect.andThen(Effect.fail(error)), + Effect.acquireUseRelease( + observeActivity( + providerSessionId, + markBusy( + providerSessionId, + busyTurnKey(input.providerThread.id, input.runOrdinal), + ), + ), + () => + runtime.startTurn(input).pipe(turnMetrics("send", input.modelSelection.model)), + (_, exit) => + Exit.isFailure(exit) + ? observeActivity( + providerSessionId, + markIdle( + providerSessionId, + busyTurnKey(input.providerThread.id, input.runOrdinal), + ), + ) + : Effect.void, ), ), ), @@ -1605,7 +1690,10 @@ export const layerWithOptions = ( return observeActivity( entry.runtime.providerSessionId, event.type === "turn.terminal" - ? markIdle(entry.runtime.providerSessionId) + ? markIdle( + entry.runtime.providerSessionId, + busyTurnKey(event.providerThreadId, event.runOrdinal), + ) : touchActivity(entry.runtime.providerSessionId), ).pipe( Effect.andThen( @@ -1695,6 +1783,12 @@ export const layerWithOptions = ( ); }; + // Parent of every session scope. On layer close, shutdown releases the + // live sessions first, then closes any session whose open is still in + // flight, time-boxed so a stuck adapter cannot hold up server shutdown. + // Parallel, so one session whose close hangs does not stop the rest from + // closing within the time box. + const sessionScopes = yield* Scope.make("parallel"); const shutdown = Effect.gen(function* () { const activeSessions = [...(yield* Ref.get(sessions)).values()]; yield* Effect.forEach( @@ -1714,7 +1808,11 @@ export const layerWithOptions = ( { discard: true }, ); }); - yield* Effect.addFinalizer(() => shutdown); + yield* Effect.addFinalizer(() => + shutdown.pipe( + Effect.ensuring(closeScopeWithin(sessionScopes, { reason: "server_shutdown" })), + ), + ); return ProviderSessionManagerV2.of({ shutdown, @@ -1783,7 +1881,7 @@ export const layerWithOptions = ( dropMcpCredentialReservation(input.threadId, mcpCredentialId); } }); - const sessionScope = yield* Scope.make(); + const sessionScope = yield* Scope.fork(sessionScopes); const runtime = yield* adapter .openSession({ threadId: input.threadId, @@ -1805,21 +1903,30 @@ export const layerWithOptions = ( }) .pipe( Effect.provideService(Scope.Scope, sessionScope), - Effect.tapError(() => - Scope.close(sessionScope, Exit.void).pipe( - Effect.ignore, - Effect.andThen(dropReservation), - // Revoke only a credential this open freshly minted: a - // reused credential is held by another live provider - // process and must survive this open's failure. + // Any failure, including a Stop that interrupts a slow + // handshake, stops the provider process this open started. + // The session cleanup runs first, and the close is + // time-boxed: this runs under the session's open lock, so an + // adapter finalizer that never finishes must not hold the + // interrupter, the lock, or later opens. + Effect.onError(() => + dropReservation.pipe( + // Clear only a session this open freshly set up: a reused + // one is held by another live provider process and must + // survive this open's failure. Effect.andThen( prepared.issued ? clearMcpSession(input.threadId, mcpCredentialId) : Effect.void, ), + Effect.ensuring( + closeScopeWithin(sessionScope, { + providerSessionId: input.providerSessionId, + reason: "open_failed", + }), + ), ), ), - Effect.onInterrupt(() => dropReservation), Effect.mapError( (cause) => new ProviderSessionOpenError({ @@ -1854,7 +1961,7 @@ export const layerWithOptions = ( requestEventPermit: yield* Semaphore.make(1), scope: sessionScope, idleGeneration: 0, - busyCount: 0, + busyTurns: new Set(), lastActivityAtMs: now, idleFiber: null, pinnedSinceMs: null, @@ -1876,12 +1983,22 @@ export const layerWithOptions = ( payload: runtime.providerSession, }), ).pipe( - Effect.tapError(() => - releaseEntry({ - providerSessionId: input.providerSessionId, - reason: "runtime_error", - detail: "Failed to persist the provider-session attachment.", - }).pipe(logReleaseFailure(input.providerSessionId)), + // Released on interrupt too: this entry has no event pump or + // idle timer yet, so nothing else would ever release it. + Effect.onError((cause) => + releaseEntry( + Cause.hasInterruptsOnly(cause) + ? { + providerSessionId: input.providerSessionId, + reason: "manual_shutdown", + detail: "The provider session start was interrupted.", + } + : { + providerSessionId: input.providerSessionId, + reason: "runtime_error", + detail: "Failed to persist the provider-session attachment.", + }, + ).pipe(logReleaseFailure(input.providerSessionId)), ), ); yield* startEventPump(entry);