From 8c0bc325ad74631143f9df73ebd0626340b9c36d Mon Sep 17 00:00:00 2001 From: tsouth89 Date: Thu, 13 Aug 2026 17:30:44 -0400 Subject: [PATCH] fix(server): keep provider notification consumers alive past startSession CursorAdapter, GrokAdapter and CodexAdapter fork their notification consumer with Effect.forkChild inside startSession. forkChild makes it a child of the calling fiber, and Effect interrupts a fiber's children when that fiber completes, so the consumer dies the moment startSession returns and every notification the provider streams afterwards is dropped. The user-visible symptom is a thread that sits on "Working" forever: the request succeeds, the provider streams its whole turn, and nothing is projected. Each adapter already builds the scope this belongs in. Cursor and Grok create a session scope, store it on the session context and close it during teardown; Codex does the same with a local sessionScope. Only the fork target was wrong. CodexSessionRuntime one layer down already uses Effect.forkIn(runtimeScope), so this makes the adapters consistent with the pattern the codebase already follows. The existing tests could not catch this. They call startSession directly from the test fiber, which never completes, so the consumer stays alive and the bug is invisible. Each new test runs startSession in a fiber it then joins, which is what production does, and asserts a notification arriving afterwards is still projected. Verified by mutation: restoring forkChild in any one adapter fails that adapter's test in about 10 seconds, and the suites pass with the fix. The tests run on the live clock. Under the default test clock the timeouts wait on virtual time that never advances, so a regression would hang until the suite timeout instead of failing where the assertion is. Fixes #5781 --- .../src/provider/Layers/CodexAdapter.test.ts | 58 ++++++++++++++++ .../src/provider/Layers/CodexAdapter.ts | 6 +- .../src/provider/Layers/CursorAdapter.test.ts | 68 +++++++++++++++++++ .../src/provider/Layers/CursorAdapter.ts | 8 ++- .../src/provider/Layers/GrokAdapter.test.ts | 67 ++++++++++++++++++ .../server/src/provider/Layers/GrokAdapter.ts | 8 ++- 6 files changed, 212 insertions(+), 3 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 7b8fbec5666..5358716aabe 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -32,6 +32,7 @@ import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import * as CodexErrors from "effect-codex-app-server/errors"; import { ServerConfig } from "../../config.ts"; @@ -1150,6 +1151,63 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { }); }), ); + + // Production calls startSession from a request fiber that finishes as soon as + // the session exists. `Effect.forkChild` made the runtime event consumer a + // child of that fiber, and Effect interrupts a fiber's children when it + // completes, so the consumer died on return and every event the session + // emitted afterwards was dropped. The other tests here start the session from + // the test fiber, which never completes, so the consumer survived and the bug + // stayed invisible. Starting it in a fiber that finishes reproduces + // production. + it.effect("keeps consuming runtime events after the startSession fiber completes", () => + Effect.gen(function* () { + const adapter = yield* CodexAdapter; + const startSessionFiber = yield* adapter + .startSession({ + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-outlives-start"), + runtimeMode: "full-access", + }) + .pipe(Effect.forkChild); + yield* Fiber.join(startSessionFiber); + + const runtime = lifecycleRuntimeFactory.lastRuntime; + NodeAssert.ok(runtime); + + const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + yield* runtime.emit({ + id: asEventId("evt-after-start-session"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + method: "item/completed", + threadId: asThreadId("thread-outlives-start"), + turnId: asTurnId("turn-1"), + itemId: asItemId("msg_after_start"), + payload: { + completedAtMs: 1_778_000_000_000, + threadId: "thread-outlives-start", + turnId: "turn-1", + item: { + type: "agentMessage", + id: "msg_after_start", + text: "emitted after startSession returned", + }, + }, + }); + + const firstEvent = yield* Fiber.join(firstEventFiber).pipe(Effect.timeout("10 seconds")); + NodeAssert.equal(firstEvent._tag, "Some"); + if (firstEvent._tag !== "Some") { + return; + } + NodeAssert.equal(firstEvent.value.type, "item.completed"); + // Live clock so the timeout above is real: under the default test clock it + // waits on virtual time that never advances, and a regression would hang + // until the suite timeout instead of failing here. + }).pipe(TestClock.withLive), + ); }); const scopedLifecycleRuntimeFactory = makeScopedRuntimeFactory(); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 6b99bf52b1e..065156d3647 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -1715,6 +1715,10 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( ), ); + // Fork into the session scope, not the calling fiber. `forkChild` makes + // this a child of `startSession`, and Effect interrupts a fiber's + // children when it completes, so the consumer died on return and every + // runtime event the session emitted afterwards was dropped. const eventFiber = yield* Stream.runForEach(runtime.events, (event) => Effect.gen(function* () { yield* writeNativeEvent(event); @@ -1730,7 +1734,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( } yield* Queue.offerAll(runtimeEventQueue, runtimeEvents); }), - ).pipe(Effect.forkChild); + ).pipe(Effect.forkIn(sessionScope)); const started = yield* runtime.start().pipe( Effect.mapError( diff --git a/apps/server/src/provider/Layers/CursorAdapter.test.ts b/apps/server/src/provider/Layers/CursorAdapter.test.ts index 491f718a977..cd5cdb7f01a 100644 --- a/apps/server/src/provider/Layers/CursorAdapter.test.ts +++ b/apps/server/src/provider/Layers/CursorAdapter.test.ts @@ -1429,4 +1429,72 @@ cursorAdapterTestLayer("CursorAdapterLive", (it) => { }).pipe(Effect.provide(customAdapterLayer)); }, ); + + // Production calls startSession from a request fiber that finishes as soon as + // the session exists. `Effect.forkChild` made the notification consumer a + // child of that fiber, and Effect interrupts a fiber's children when it + // completes, so the consumer died on return and every later session/update + // was dropped: the thread sat on "Working" forever while the provider + // streamed its whole turn. The other tests here call startSession directly + // from the test fiber, which never completes, so the consumer survived and + // the bug stayed invisible. Running it in a fiber that finishes is what + // reproduces production. + it.effect("keeps consuming notifications after the startSession fiber completes", () => + Effect.gen(function* () { + const adapter = yield* CursorAdapter; + const settings = yield* ServerSettingsService; + const threadId = ThreadId.make("cursor-consumer-outlives-start-session"); + + const wrapperPath = yield* Effect.promise(() => makeMockAgentWrapper()); + yield* settings.updateSettings({ providers: { cursor: { binaryPath: wrapperPath } } }); + + const runtimeEvents: ProviderRuntimeEvent[] = []; + const sawContentDelta = yield* Deferred.make(); + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.sync(() => { + runtimeEvents.push(event); + }).pipe( + Effect.andThen( + event.type === "content.delta" && String(event.threadId) === String(threadId) + ? Deferred.succeed(sawContentDelta, undefined).pipe(Effect.asVoid) + : Effect.void, + ), + ), + ).pipe(Effect.forkChild); + + const startSessionFiber = yield* adapter + .startSession({ + threadId, + provider: ProviderDriverKind.make("cursor"), + cwd: process.cwd(), + runtimeMode: "full-access", + modelSelection: { instanceId: ProviderInstanceId.make("cursor"), model: "default" }, + }) + .pipe(Effect.forkChild); + yield* Fiber.join(startSessionFiber).pipe(Effect.timeout("10 seconds")); + + // Forked, and the assertion waits on the projected event rather than on + // sendTurn: with the consumer dead the turn never settles, so awaiting it + // directly would hang until the suite timeout instead of failing here. + const sendTurnFiber = yield* adapter + .sendTurn({ threadId, input: "hello mock", attachments: [] }) + .pipe(Effect.forkChild); + yield* Deferred.await(sawContentDelta).pipe(Effect.timeout("10 seconds")); + yield* Fiber.join(sendTurnFiber).pipe(Effect.timeout("10 seconds")); + + const delta = runtimeEvents.find( + (event) => event.type === "content.delta" && String(event.threadId) === String(threadId), + ); + assert.isDefined( + delta, + "no content.delta was projected after the startSession fiber completed", + ); + + yield* Fiber.interrupt(runtimeEventsFiber); + yield* adapter.stopSession(threadId); + // Live clock so the timeouts above are real: under the default test clock + // they wait on virtual time that never advances, and a regression would + // hang until the suite timeout instead of failing here. + }).pipe(TestClock.withLive), + ); }); diff --git a/apps/server/src/provider/Layers/CursorAdapter.ts b/apps/server/src/provider/Layers/CursorAdapter.ts index 80475a5c269..30c173d8fae 100644 --- a/apps/server/src/provider/Layers/CursorAdapter.ts +++ b/apps/server/src/provider/Layers/CursorAdapter.ts @@ -874,7 +874,13 @@ export function makeCursorAdapter( Effect.catch((cause) => Effect.logError("Failed to process Cursor runtime notification.", { cause }), ), - Effect.forkChild, + // Fork into the session scope, not the calling fiber. `forkChild` + // makes this a child of `startSession`, and Effect interrupts a + // fiber's children when it completes, so the consumer died as soon + // as `startSession` returned and every later notification was + // dropped. The scope is created, stored on the context and closed + // on teardown already; only the fork target was wrong. + Effect.forkIn(ctx.scope), ); ctx.notificationFiber = nf; diff --git a/apps/server/src/provider/Layers/GrokAdapter.test.ts b/apps/server/src/provider/Layers/GrokAdapter.test.ts index 7b6f0972ae8..6cb71660a74 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.test.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.test.ts @@ -1197,4 +1197,71 @@ it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => { yield* adapter.stopSession(threadId); }), ); + + // Production calls startSession from a request fiber that finishes as soon as + // the session exists. `Effect.forkChild` made the notification consumer a + // child of that fiber, and Effect interrupts a fiber's children when it + // completes, so the consumer died on return and every later session/update + // was dropped: the thread sat on "Working" forever while the provider + // streamed its whole turn. Every other test here calls startSession directly + // from the test fiber, which never completes, so the consumer survived and + // the bug stayed invisible. Running it in a fiber that finishes is what + // reproduces production. + it.effect("keeps consuming notifications after the startSession fiber completes", () => + Effect.gen(function* () { + const threadId = ThreadId.make("grok-consumer-outlives-start-session"); + const wrapperPath = yield* Effect.promise(() => makeMockGrokWrapper()); + const adapter = yield* makeTestAdapter(wrapperPath); + + const runtimeEvents: ProviderRuntimeEvent[] = []; + const turnCompleted = yield* Deferred.make(); + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.sync(() => { + runtimeEvents.push(event); + }).pipe( + Effect.andThen( + event.type === "turn.completed" && String(event.threadId) === String(threadId) + ? Deferred.succeed(turnCompleted, undefined).pipe(Effect.asVoid) + : Effect.void, + ), + ), + ).pipe(Effect.forkChild); + + const startSessionFiber = yield* adapter + .startSession({ + threadId, + provider: ProviderDriverKind.make("grok"), + cwd: process.cwd(), + runtimeMode: "full-access", + }) + .pipe(Effect.forkChild); + yield* Fiber.join(startSessionFiber).pipe(Effect.timeout("10 seconds")); + + // Forked, and the assertion waits on the projected event rather than on + // sendTurn: with the consumer dead the turn never settles, so awaiting it + // directly would hang until the suite timeout instead of failing here. + const sendTurnFiber = yield* adapter + .sendTurn({ threadId, input: "hello grok", attachments: [] }) + .pipe(Effect.forkChild); + yield* Deferred.await(turnCompleted).pipe(Effect.timeout("10 seconds")); + yield* Fiber.join(sendTurnFiber).pipe(Effect.timeout("10 seconds")); + + const delta = runtimeEvents.find( + (event) => event.type === "content.delta" && String(event.threadId) === String(threadId), + ); + assert.isDefined( + delta, + "no content.delta was projected after the startSession fiber completed", + ); + if (delta?.type === "content.delta") { + assert.equal(delta.payload.delta, "hello from mock"); + } + + yield* Fiber.interrupt(runtimeEventsFiber); + yield* adapter.stopSession(threadId); + // Live clock so the timeouts above are real: under the default test clock + // they wait on virtual time that never advances, and a regression would + // hang until the suite timeout instead of failing here. + }).pipe(TestClock.withLive), + ); }); diff --git a/apps/server/src/provider/Layers/GrokAdapter.ts b/apps/server/src/provider/Layers/GrokAdapter.ts index 977cc8caadd..858d862e6d5 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.ts @@ -876,7 +876,13 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte Effect.catch((cause) => Effect.logError("Failed to process Grok runtime notification.", { cause }), ), - Effect.forkChild, + // Fork into the session scope, not the calling fiber. `forkChild` + // makes this a child of `startSession`, and Effect interrupts a + // fiber's children when it completes, so the consumer died as soon + // as `startSession` returned and every later notification was + // dropped. The scope is created, stored on the context and closed + // on teardown already; only the fork target was wrong. + Effect.forkIn(ctx.scope), ); ctx.notificationFiber = nf;