authorgravatar for 35393046+lnieuwenhuis@users.noreply.github.comLars Nieuwenhuis <35393046+lnieuwenhuis@users.noreply.github.com> 2026-09-15 22:52:54+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-15 13:52:54-07:00
loga5da32750125c3bedb959bf331e5e49e95b95e5d
tree6a62362126c817bbcde2ab63aea000b13563ff53
parent3efdcc5296f1754e0f3bf7fee5fc2ada510e0438
signature Signed by PGP key B5690EEEBB952194

fix(client-runtime): preserve cached turns and older-page loading (#8309)

Co-authored-by: Julius Marminge <julius0216@outlook.com>

3 files changed, 177 insertions(+), 23 deletions(-)

packages/client-runtime/src/state/threads-pagination.test.ts+59
......@@ -398,6 +398,65 @@ describe("thread pagination state", () => {
398398 }),
399399 );
400400
401 it.effect("keeps a new page loading when a snapshot replaced a parked older page", () =>
402 Effect.gen(function* () {
403 const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) });
404 yield* harness.awaitState((value) => Option.isSome(value.page));
405 requestOlderThreadTurns(TARGET.environmentId, THREAD_ID);
406 yield* harness.resolveNextPage(
407 Option.some({
408 ...OLDER_PAGE,
409 snapshotSequence: 30,
410 page: { beforeCursor: null, hasMore: false, snapshotSequence: 30, threadSequence: 30 },
411 }),
412 );
413 yield* Queue.offer(harness.inputs, titleEvent("Waiting for old watermark", 11));
414 yield* harness.awaitState((value) =>
415 Option.exists(value.data, (thread) => thread.title === "Waiting for old watermark"),
416 );
417 expect(
418 Option.getOrThrow((yield* SubscriptionRef.get(harness.threadState)).page).loadingOlder,
419 ).toBe(true);
420
421 yield* Queue.offer(harness.inputs, {
422 kind: "snapshot",
423 snapshot: {
424 snapshotSequence: 20,
425 thread: { ...BASE_THREAD, title: "Replacement snapshot" },
426 page: { beforeCursor: "cursor-2", hasMore: true, snapshotSequence: 20 },
427 },
428 });
429 yield* harness.awaitState((value) =>
430 Option.exists(value.data, (thread) => thread.title === "Replacement snapshot"),
431 );
432 requestOlderThreadTurns(TARGET.environmentId, THREAD_ID);
433 yield* harness.awaitState((value) =>
434 Option.exists(value.page, (page) => page.loadingOlder && page.beforeCursor === "cursor-2"),
435 );
436 yield* Queue.offer(harness.inputs, titleEvent("New request still loading", 21));
437 yield* harness.awaitState((value) =>
438 Option.exists(value.data, (thread) => thread.title === "New request still loading"),
439 );
440 const loading = yield* SubscriptionRef.get(harness.threadState);
441 expect(Option.getOrThrow(loading.page).loadingOlder).toBe(true);
442 expect(hasMessage(loading, "message-old")).toBe(false);
443 expect((yield* Ref.get(harness.loaderWindows)).map((window) => window?.beforeCursor)).toEqual(
444 [undefined, "cursor-1", "cursor-2"],
445 );
446
447 yield* harness.resolveNextPage(
448 Option.some({
449 ...OLDER_PAGE,
450 snapshotSequence: 21,
451 page: { beforeCursor: null, hasMore: false, snapshotSequence: 21, threadSequence: 21 },
452 }),
453 );
454 const completed = yield* harness.awaitState((value) => hasMessage(value, "message-old"));
455 expect(Option.getOrThrow(completed.page).loadingOlder).toBe(false);
456 expect(Option.getOrThrow(completed.page).beforeCursor).toBeNull();
457 }),
458 );
459
401460 it.effect("discards an older page read from a projection behind the loaded state", () =>
402461 Effect.gen(function* () {
403462 const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) });
packages/client-runtime/src/state/threads-sync.test.ts+81-4
......@@ -146,6 +146,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
146146 const inputs = yield* Queue.unbounded<TestThreadInput>();
147147 const observed = yield* Queue.unbounded<EnvironmentThreadState>();
148148 const latest = yield* Ref.make<EnvironmentThreadState>(EMPTY_ENVIRONMENT_THREAD_STATE);
149 const stateChangeCount = yield* Ref.make(0);
149150 const retryCount = yield* Ref.make(0);
150151 const subscriptionCount = yield* Ref.make(0);
151152 const loaderCalls = yield* Ref.make(0);
......@@ -157,11 +158,19 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
157158 const supervisorState = yield* SubscriptionRef.make<SupervisorConnectionState>(
158159 AVAILABLE_CONNECTION_STATE,
159160 );
161 // Preserve queued event batches while failing at the first error.
160162 const streamFrom = (queue: Queue.Queue<TestThreadInput>) =>
161163 Stream.fromQueue(queue).pipe(
162 Stream.mapEffect((input) =>
163 input instanceof Error ? Effect.fail(input) : Effect.succeed(input),
164 ),
164 Stream.chunks,
165 Stream.flatMap((chunk) => {
166 const errorIndex = chunk.findIndex((input) => input instanceof Error);
167 if (errorIndex === -1) {
168 return Stream.fromArray(chunk as ReadonlyArray<OrchestrationThreadStreamItem>);
169 }
170 const prefix = chunk.slice(0, errorIndex) as ReadonlyArray<OrchestrationThreadStreamItem>;
171 const failure = Stream.fail(chunk[errorIndex] as Error);
172 return prefix.length === 0 ? failure : Stream.concat(Stream.fromArray(prefix), failure);
173 }),
165174 );
166175 const client = {
167176 [ORCHESTRATION_WS_METHODS.subscribeThread]: (input: {
......@@ -244,7 +253,10 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
244253 );
245254 yield* SubscriptionRef.changes(threadState).pipe(
246255 Stream.runForEach((state) =>
247 Ref.set(latest, state).pipe(Effect.andThen(Queue.offer(observed, state))),
256 Ref.update(stateChangeCount, (count) => count + 1).pipe(
257 Effect.andThen(Ref.set(latest, state)),
258 Effect.andThen(Queue.offer(observed, state)),
259 ),
248260 ),
249261 Effect.forkScoped,
250262 );
......@@ -254,6 +266,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
254266 inputs,
255267 observed,
256268 latest,
269 stateChangeCount,
257270 retryCount,
258271 subscriptionCount,
259272 loaderCalls,
......@@ -307,6 +320,38 @@ const titleUpdated = (title: string, sequence = 2): OrchestrationThreadStreamIte
307320 },
308321});
309322
323const sessionSet = (
324 status: "ready" | "running",
325 turnId: string,
326 sequence: number,
327): OrchestrationThreadStreamItem => ({
328 kind: "event",
329 event: {
330 eventId: EventId.make(`event-session-${status}-${sequence}`),
331 sequence,
332 occurredAt: "2026-04-01T03:00:00.000Z",
333 commandId: null,
334 causationEventId: null,
335 correlationId: null,
336 metadata: {},
337 aggregateKind: "thread",
338 aggregateId: THREAD_ID,
339 type: "thread.session-set",
340 payload: {
341 threadId: THREAD_ID,
342 session: {
343 threadId: THREAD_ID,
344 status,
345 providerName: "codex",
346 runtimeMode: "full-access",
347 activeTurnId: status === "running" ? TurnId.make(turnId) : null,
348 lastError: null,
349 updatedAt: "2026-04-01T03:00:00.000Z",
350 },
351 },
352 },
353});
354
310355const deleted = (): OrchestrationThreadStreamItem => ({
311356 kind: "event",
312357 event: {
......@@ -989,4 +1034,36 @@ describe("EnvironmentThreads", () => {
9891034 expect(yield* Ref.get(harness.subscriptionCount)).toBe(3);
9901035 }),
9911036 );
1037
1038 it.effect(
1039 "persists a turn that settles mid-batch when the next turn starts in the same batch",
1040 () =>
1041 Effect.gen(function* () {
1042 const harness = yield* makeHarness({ cached: ACTIVE_THREAD });
1043 yield* awaitThreadState(harness.observed, (value) => value.status === "live");
1044 const before = yield* Ref.get(harness.stateChangeCount);
1045
1046 // Both events arrive in one transport batch: the session settles and the
1047 // next turn starts before the fold publishes.
1048 yield* Queue.offerAll(harness.inputs, [
1049 sessionSet("ready", "turn-1", CACHED_SNAPSHOT_SEQUENCE + 1),
1050 sessionSet("running", "turn-2", CACHED_SNAPSHOT_SEQUENCE + 2),
1051 ]);
1052 yield* awaitThreadState(
1053 harness.observed,
1054 (value) =>
1055 Option.isSome(value.data) &&
1056 value.data.value.session?.activeTurnId === TurnId.make("turn-2"),
1057 );
1058 expect((yield* Ref.get(harness.stateChangeCount)) - before).toBe(1);
1059 yield* TestClock.adjust("500 millis");
1060 yield* Effect.yieldNow;
1061
1062 // The settled state reached the cache under its own sequence even
1063 // though the batch ended on a running session.
1064 const saved = (yield* Ref.get(harness.savedThreads)).at(-1);
1065 expect(saved?.thread.session?.status).toBe("ready");
1066 expect(saved?.snapshotSequence).toBe(CACHED_SNAPSHOT_SEQUENCE + 1);
1067 }),
1068 );
9921069});
packages/client-runtime/src/state/threads.ts+37-19
......@@ -353,6 +353,29 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
353353 ),
354354 );
355355
356 const offerThreadPersistence = Effect.fn("EnvironmentThreadState.offerThreadPersistence")(
357 function* (thread: OrchestrationThread, snapshotSequence: number) {
358 const currentPage = yield* SubscriptionRef.get(state).pipe(Effect.map((value) => value.page));
359 yield* Queue.offer(persistence, {
360 snapshotSequence,
361 thread,
362 // Persist the window boundary with the window's content so a cache
363 // restore can keep paging from where the loaded history ends.
364 ...Option.match(currentPage, {
365 onNone: () => ({}),
366 onSome: (value) =>
367 ({
368 page: {
369 beforeCursor: value.beforeCursor,
370 hasMore: value.hasMore,
371 snapshotSequence,
372 },
373 }) as const,
374 }),
375 });
376 },
377 );
378
356379 const setThread = Effect.fn("EnvironmentThreadState.setThread")(function* (
357380 thread: OrchestrationThread,
358381 // "keep" preserves the current page state (live events touch only loaded
......@@ -376,24 +399,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
376399 // persist once it settles so cache encoding stays off the streaming path.
377400 if (shouldPersistThread(thread)) {
378401 const snapshotSequence = yield* SubscriptionRef.get(lastSequence);
379 const currentPage = yield* SubscriptionRef.get(state).pipe(Effect.map((value) => value.page));
380 yield* Queue.offer(persistence, {
381 snapshotSequence,
382 thread,
383 // Persist the window boundary with the window's content so a cache
384 // restore can keep paging from where the loaded history ends.
385 ...Option.match(currentPage, {
386 onNone: () => ({}),
387 onSome: (value) =>
388 ({
389 page: {
390 beforeCursor: value.beforeCursor,
391 hasMore: value.hasMore,
392 snapshotSequence,
393 },
394 }) as const,
395 }),
396 });
402 yield* offerThreadPersistence(thread, snapshotSequence);
397403 }
398404 });
399405
......@@ -441,6 +447,9 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
441447 // in the preserved history with no event left to remove it. The
442448 // epoch bump discards any older-page fetch racing this snapshot.
443449 yield* Ref.update(historyEpoch, (epoch) => epoch + 1);
450 // A parked response must not clear loadingOlder on a request started
451 // from the replacement snapshot's cursor.
452 yield* Ref.set(pendingOlderPage, null);
444453 yield* SubscriptionRef.set(lastSequence, item.snapshot.snapshotSequence);
445454 yield* setThread(item.snapshot.thread, pageStateFromSnapshot(item.snapshot.page));
446455 return;
......@@ -539,17 +548,26 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
539548 let thread = current.data.value;
540549 let sequence = yield* SubscriptionRef.get(lastSequence);
541550 let synchronized = false;
551 // Retain the last settled state even if the next turn starts before
552 // this batch publishes. Its cursor must describe that settled content.
553 let persistable: { thread: OrchestrationThread; sequence: number } | undefined;
542554 for (const item of items) {
543555 if (item.kind === "synchronized") {
544556 synchronized = true;
545557 } else if (item.kind === "event" && item.event.sequence > sequence) {
546558 sequence = item.event.sequence;
547559 const result = applyThreadDetailEvent(thread, item.event);
548 if (result.kind === "updated") thread = result.thread;
560 if (result.kind === "updated") {
561 thread = result.thread;
562 if (shouldPersistThread(thread)) persistable = { thread, sequence };
563 }
549564 }
550565 }
551566 yield* SubscriptionRef.set(lastSequence, sequence);
552567 if (thread !== current.data.value) yield* setThread(thread, "keep");
568 if (persistable !== undefined && !shouldPersistThread(thread)) {
569 yield* offerThreadPersistence(persistable.thread, persistable.sequence);
570 }
553571 if (synchronized) yield* applyItemLocked({ kind: "synchronized" });
554572 yield* remember;
555573 }),