fix(server): keep provider notification consumers alive past startSession (#6538)

Co-authored-by: tsouth89 <tsouth89@users.noreply.github.com>

Author: tsouth89

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();

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(

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<void>();
+      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),
+  );
 });

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;

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<void>();
+      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),
+  );
 });

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;