authorgravatar for tyler@southboundsoftware.comTyler <tyler@southboundsoftware.com> 2026-08-15 05:43:13-05:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-08-15 12:43:13+02:00
logafca73d3683c99057ea8af1ad7d77511a0faf680
tree31677c6dac880a3ac0641466af097c9b982778c6
parentd8a6dfd31539a86d08bd4fbd030f8252b3c405ac
signature Signed by PGP key B5690EEEBB952194

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

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

6 files changed, 212 insertions(+), 3 deletions(-)

apps/server/src/provider/Layers/CodexAdapter.test.ts+58
......@@ -32,6 +32,7 @@ import * as Queue from "effect/Queue";
3232import * as Schema from "effect/Schema";
3333import * as Scope from "effect/Scope";
3434import * as Stream from "effect/Stream";
35import * as TestClock from "effect/testing/TestClock";
3536import * as CodexErrors from "effect-codex-app-server/errors";
3637
3738import { ServerConfig } from "../../config.ts";
......@@ -1150,6 +1151,63 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => {
11501151 });
11511152 }),
11521153 );
1154
1155 // Production calls startSession from a request fiber that finishes as soon as
1156 // the session exists. `Effect.forkChild` made the runtime event consumer a
1157 // child of that fiber, and Effect interrupts a fiber's children when it
1158 // completes, so the consumer died on return and every event the session
1159 // emitted afterwards was dropped. The other tests here start the session from
1160 // the test fiber, which never completes, so the consumer survived and the bug
1161 // stayed invisible. Starting it in a fiber that finishes reproduces
1162 // production.
1163 it.effect("keeps consuming runtime events after the startSession fiber completes", () =>
1164 Effect.gen(function* () {
1165 const adapter = yield* CodexAdapter;
1166 const startSessionFiber = yield* adapter
1167 .startSession({
1168 provider: ProviderDriverKind.make("codex"),
1169 threadId: asThreadId("thread-outlives-start"),
1170 runtimeMode: "full-access",
1171 })
1172 .pipe(Effect.forkChild);
1173 yield* Fiber.join(startSessionFiber);
1174
1175 const runtime = lifecycleRuntimeFactory.lastRuntime;
1176 NodeAssert.ok(runtime);
1177
1178 const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild);
1179 yield* runtime.emit({
1180 id: asEventId("evt-after-start-session"),
1181 kind: "notification",
1182 provider: ProviderDriverKind.make("codex"),
1183 createdAt: "2026-01-01T00:00:00.000Z",
1184 method: "item/completed",
1185 threadId: asThreadId("thread-outlives-start"),
1186 turnId: asTurnId("turn-1"),
1187 itemId: asItemId("msg_after_start"),
1188 payload: {
1189 completedAtMs: 1_778_000_000_000,
1190 threadId: "thread-outlives-start",
1191 turnId: "turn-1",
1192 item: {
1193 type: "agentMessage",
1194 id: "msg_after_start",
1195 text: "emitted after startSession returned",
1196 },
1197 },
1198 });
1199
1200 const firstEvent = yield* Fiber.join(firstEventFiber).pipe(Effect.timeout("10 seconds"));
1201 NodeAssert.equal(firstEvent._tag, "Some");
1202 if (firstEvent._tag !== "Some") {
1203 return;
1204 }
1205 NodeAssert.equal(firstEvent.value.type, "item.completed");
1206 // Live clock so the timeout above is real: under the default test clock it
1207 // waits on virtual time that never advances, and a regression would hang
1208 // until the suite timeout instead of failing here.
1209 }).pipe(TestClock.withLive),
1210 );
11531211});
11541212
11551213const scopedLifecycleRuntimeFactory = makeScopedRuntimeFactory();
apps/server/src/provider/Layers/CodexAdapter.ts+5-1
......@@ -1715,6 +1715,10 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
17151715 ),
17161716 );
17171717
1718 // Fork into the session scope, not the calling fiber. `forkChild` makes
1719 // this a child of `startSession`, and Effect interrupts a fiber's
1720 // children when it completes, so the consumer died on return and every
1721 // runtime event the session emitted afterwards was dropped.
17181722 const eventFiber = yield* Stream.runForEach(runtime.events, (event) =>
17191723 Effect.gen(function* () {
17201724 yield* writeNativeEvent(event);
......@@ -1730,7 +1734,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
17301734 }
17311735 yield* Queue.offerAll(runtimeEventQueue, runtimeEvents);
17321736 }),
1733 ).pipe(Effect.forkChild);
1737 ).pipe(Effect.forkIn(sessionScope));
17341738
17351739 const started = yield* runtime.start().pipe(
17361740 Effect.mapError(
apps/server/src/provider/Layers/CursorAdapter.test.ts+68
......@@ -1429,4 +1429,72 @@ cursorAdapterTestLayer("CursorAdapterLive", (it) => {
14291429 }).pipe(Effect.provide(customAdapterLayer));
14301430 },
14311431 );
1432
1433 // Production calls startSession from a request fiber that finishes as soon as
1434 // the session exists. `Effect.forkChild` made the notification consumer a
1435 // child of that fiber, and Effect interrupts a fiber's children when it
1436 // completes, so the consumer died on return and every later session/update
1437 // was dropped: the thread sat on "Working" forever while the provider
1438 // streamed its whole turn. The other tests here call startSession directly
1439 // from the test fiber, which never completes, so the consumer survived and
1440 // the bug stayed invisible. Running it in a fiber that finishes is what
1441 // reproduces production.
1442 it.effect("keeps consuming notifications after the startSession fiber completes", () =>
1443 Effect.gen(function* () {
1444 const adapter = yield* CursorAdapter;
1445 const settings = yield* ServerSettingsService;
1446 const threadId = ThreadId.make("cursor-consumer-outlives-start-session");
1447
1448 const wrapperPath = yield* Effect.promise(() => makeMockAgentWrapper());
1449 yield* settings.updateSettings({ providers: { cursor: { binaryPath: wrapperPath } } });
1450
1451 const runtimeEvents: ProviderRuntimeEvent[] = [];
1452 const sawContentDelta = yield* Deferred.make<void>();
1453 const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
1454 Effect.sync(() => {
1455 runtimeEvents.push(event);
1456 }).pipe(
1457 Effect.andThen(
1458 event.type === "content.delta" && String(event.threadId) === String(threadId)
1459 ? Deferred.succeed(sawContentDelta, undefined).pipe(Effect.asVoid)
1460 : Effect.void,
1461 ),
1462 ),
1463 ).pipe(Effect.forkChild);
1464
1465 const startSessionFiber = yield* adapter
1466 .startSession({
1467 threadId,
1468 provider: ProviderDriverKind.make("cursor"),
1469 cwd: process.cwd(),
1470 runtimeMode: "full-access",
1471 modelSelection: { instanceId: ProviderInstanceId.make("cursor"), model: "default" },
1472 })
1473 .pipe(Effect.forkChild);
1474 yield* Fiber.join(startSessionFiber).pipe(Effect.timeout("10 seconds"));
1475
1476 // Forked, and the assertion waits on the projected event rather than on
1477 // sendTurn: with the consumer dead the turn never settles, so awaiting it
1478 // directly would hang until the suite timeout instead of failing here.
1479 const sendTurnFiber = yield* adapter
1480 .sendTurn({ threadId, input: "hello mock", attachments: [] })
1481 .pipe(Effect.forkChild);
1482 yield* Deferred.await(sawContentDelta).pipe(Effect.timeout("10 seconds"));
1483 yield* Fiber.join(sendTurnFiber).pipe(Effect.timeout("10 seconds"));
1484
1485 const delta = runtimeEvents.find(
1486 (event) => event.type === "content.delta" && String(event.threadId) === String(threadId),
1487 );
1488 assert.isDefined(
1489 delta,
1490 "no content.delta was projected after the startSession fiber completed",
1491 );
1492
1493 yield* Fiber.interrupt(runtimeEventsFiber);
1494 yield* adapter.stopSession(threadId);
1495 // Live clock so the timeouts above are real: under the default test clock
1496 // they wait on virtual time that never advances, and a regression would
1497 // hang until the suite timeout instead of failing here.
1498 }).pipe(TestClock.withLive),
1499 );
14321500});
apps/server/src/provider/Layers/CursorAdapter.ts+7-1
......@@ -874,7 +874,13 @@ export function makeCursorAdapter(
874874 Effect.catch((cause) =>
875875 Effect.logError("Failed to process Cursor runtime notification.", { cause }),
876876 ),
877 Effect.forkChild,
877 // Fork into the session scope, not the calling fiber. `forkChild`
878 // makes this a child of `startSession`, and Effect interrupts a
879 // fiber's children when it completes, so the consumer died as soon
880 // as `startSession` returned and every later notification was
881 // dropped. The scope is created, stored on the context and closed
882 // on teardown already; only the fork target was wrong.
883 Effect.forkIn(ctx.scope),
878884 );
879885
880886 ctx.notificationFiber = nf;
apps/server/src/provider/Layers/GrokAdapter.test.ts+67
......@@ -1197,4 +1197,71 @@ it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => {
11971197 yield* adapter.stopSession(threadId);
11981198 }),
11991199 );
1200
1201 // Production calls startSession from a request fiber that finishes as soon as
1202 // the session exists. `Effect.forkChild` made the notification consumer a
1203 // child of that fiber, and Effect interrupts a fiber's children when it
1204 // completes, so the consumer died on return and every later session/update
1205 // was dropped: the thread sat on "Working" forever while the provider
1206 // streamed its whole turn. Every other test here calls startSession directly
1207 // from the test fiber, which never completes, so the consumer survived and
1208 // the bug stayed invisible. Running it in a fiber that finishes is what
1209 // reproduces production.
1210 it.effect("keeps consuming notifications after the startSession fiber completes", () =>
1211 Effect.gen(function* () {
1212 const threadId = ThreadId.make("grok-consumer-outlives-start-session");
1213 const wrapperPath = yield* Effect.promise(() => makeMockGrokWrapper());
1214 const adapter = yield* makeTestAdapter(wrapperPath);
1215
1216 const runtimeEvents: ProviderRuntimeEvent[] = [];
1217 const turnCompleted = yield* Deferred.make<void>();
1218 const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
1219 Effect.sync(() => {
1220 runtimeEvents.push(event);
1221 }).pipe(
1222 Effect.andThen(
1223 event.type === "turn.completed" && String(event.threadId) === String(threadId)
1224 ? Deferred.succeed(turnCompleted, undefined).pipe(Effect.asVoid)
1225 : Effect.void,
1226 ),
1227 ),
1228 ).pipe(Effect.forkChild);
1229
1230 const startSessionFiber = yield* adapter
1231 .startSession({
1232 threadId,
1233 provider: ProviderDriverKind.make("grok"),
1234 cwd: process.cwd(),
1235 runtimeMode: "full-access",
1236 })
1237 .pipe(Effect.forkChild);
1238 yield* Fiber.join(startSessionFiber).pipe(Effect.timeout("10 seconds"));
1239
1240 // Forked, and the assertion waits on the projected event rather than on
1241 // sendTurn: with the consumer dead the turn never settles, so awaiting it
1242 // directly would hang until the suite timeout instead of failing here.
1243 const sendTurnFiber = yield* adapter
1244 .sendTurn({ threadId, input: "hello grok", attachments: [] })
1245 .pipe(Effect.forkChild);
1246 yield* Deferred.await(turnCompleted).pipe(Effect.timeout("10 seconds"));
1247 yield* Fiber.join(sendTurnFiber).pipe(Effect.timeout("10 seconds"));
1248
1249 const delta = runtimeEvents.find(
1250 (event) => event.type === "content.delta" && String(event.threadId) === String(threadId),
1251 );
1252 assert.isDefined(
1253 delta,
1254 "no content.delta was projected after the startSession fiber completed",
1255 );
1256 if (delta?.type === "content.delta") {
1257 assert.equal(delta.payload.delta, "hello from mock");
1258 }
1259
1260 yield* Fiber.interrupt(runtimeEventsFiber);
1261 yield* adapter.stopSession(threadId);
1262 // Live clock so the timeouts above are real: under the default test clock
1263 // they wait on virtual time that never advances, and a regression would
1264 // hang until the suite timeout instead of failing here.
1265 }).pipe(TestClock.withLive),
1266 );
12001267});
apps/server/src/provider/Layers/GrokAdapter.ts+7-1
......@@ -876,7 +876,13 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
876876 Effect.catch((cause) =>
877877 Effect.logError("Failed to process Grok runtime notification.", { cause }),
878878 ),
879 Effect.forkChild,
879 // Fork into the session scope, not the calling fiber. `forkChild`
880 // makes this a child of `startSession`, and Effect interrupts a
881 // fiber's children when it completes, so the consumer died as soon
882 // as `startSession` returned and every later notification was
883 // dropped. The scope is created, stored on the context and closed
884 // on teardown already; only the fork target was wrong.
885 Effect.forkIn(ctx.scope),
880886 );
881887
882888 ctx.notificationFiber = nf;