From a5da32750125c3bedb959bf331e5e49e95b95e5d Mon Sep 17 00:00:00 2001 From: Lars Nieuwenhuis <35393046+lnieuwenhuis@users.noreply.github.com> Date: Tue, 15 Sep 2026 22:52:54 +0200 Subject: [PATCH] fix(client-runtime): preserve cached turns and older-page loading (#8309) Co-authored-by: Julius Marminge --- .../src/state/threads-pagination.test.ts | 59 +++++++++++++ .../src/state/threads-sync.test.ts | 85 ++++++++++++++++++- packages/client-runtime/src/state/threads.ts | 56 +++++++----- 3 files changed, 177 insertions(+), 23 deletions(-) diff --git a/packages/client-runtime/src/state/threads-pagination.test.ts b/packages/client-runtime/src/state/threads-pagination.test.ts index 41e80666fd84d82ad50f2a1e10ed9f1bbbb2c96f..a583908e846ce0ef095733b34d004f2c7b62095d 100644 --- a/packages/client-runtime/src/state/threads-pagination.test.ts +++ b/packages/client-runtime/src/state/threads-pagination.test.ts @@ -398,6 +398,65 @@ describe("thread pagination state", () => { }), ); + it.effect("keeps a new page loading when a snapshot replaced a parked older page", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) }); + yield* harness.awaitState((value) => Option.isSome(value.page)); + requestOlderThreadTurns(TARGET.environmentId, THREAD_ID); + yield* harness.resolveNextPage( + Option.some({ + ...OLDER_PAGE, + snapshotSequence: 30, + page: { beforeCursor: null, hasMore: false, snapshotSequence: 30, threadSequence: 30 }, + }), + ); + yield* Queue.offer(harness.inputs, titleEvent("Waiting for old watermark", 11)); + yield* harness.awaitState((value) => + Option.exists(value.data, (thread) => thread.title === "Waiting for old watermark"), + ); + expect( + Option.getOrThrow((yield* SubscriptionRef.get(harness.threadState)).page).loadingOlder, + ).toBe(true); + + yield* Queue.offer(harness.inputs, { + kind: "snapshot", + snapshot: { + snapshotSequence: 20, + thread: { ...BASE_THREAD, title: "Replacement snapshot" }, + page: { beforeCursor: "cursor-2", hasMore: true, snapshotSequence: 20 }, + }, + }); + yield* harness.awaitState((value) => + Option.exists(value.data, (thread) => thread.title === "Replacement snapshot"), + ); + requestOlderThreadTurns(TARGET.environmentId, THREAD_ID); + yield* harness.awaitState((value) => + Option.exists(value.page, (page) => page.loadingOlder && page.beforeCursor === "cursor-2"), + ); + yield* Queue.offer(harness.inputs, titleEvent("New request still loading", 21)); + yield* harness.awaitState((value) => + Option.exists(value.data, (thread) => thread.title === "New request still loading"), + ); + const loading = yield* SubscriptionRef.get(harness.threadState); + expect(Option.getOrThrow(loading.page).loadingOlder).toBe(true); + expect(hasMessage(loading, "message-old")).toBe(false); + expect((yield* Ref.get(harness.loaderWindows)).map((window) => window?.beforeCursor)).toEqual( + [undefined, "cursor-1", "cursor-2"], + ); + + yield* harness.resolveNextPage( + Option.some({ + ...OLDER_PAGE, + snapshotSequence: 21, + page: { beforeCursor: null, hasMore: false, snapshotSequence: 21, threadSequence: 21 }, + }), + ); + const completed = yield* harness.awaitState((value) => hasMessage(value, "message-old")); + expect(Option.getOrThrow(completed.page).loadingOlder).toBe(false); + expect(Option.getOrThrow(completed.page).beforeCursor).toBeNull(); + }), + ); + it.effect("discards an older page read from a projection behind the loaded state", () => Effect.gen(function* () { const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) }); diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index ef09c92aa8e979faff4f570445752d7b7f3fe620..c41651576dad363ebb973fa8d6355278d6f81a87 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -146,6 +146,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o const inputs = yield* Queue.unbounded(); const observed = yield* Queue.unbounded(); const latest = yield* Ref.make(EMPTY_ENVIRONMENT_THREAD_STATE); + const stateChangeCount = yield* Ref.make(0); const retryCount = yield* Ref.make(0); const subscriptionCount = yield* Ref.make(0); const loaderCalls = yield* Ref.make(0); @@ -157,11 +158,19 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o const supervisorState = yield* SubscriptionRef.make( AVAILABLE_CONNECTION_STATE, ); + // Preserve queued event batches while failing at the first error. const streamFrom = (queue: Queue.Queue) => Stream.fromQueue(queue).pipe( - Stream.mapEffect((input) => - input instanceof Error ? Effect.fail(input) : Effect.succeed(input), - ), + Stream.chunks, + Stream.flatMap((chunk) => { + const errorIndex = chunk.findIndex((input) => input instanceof Error); + if (errorIndex === -1) { + return Stream.fromArray(chunk as ReadonlyArray); + } + const prefix = chunk.slice(0, errorIndex) as ReadonlyArray; + const failure = Stream.fail(chunk[errorIndex] as Error); + return prefix.length === 0 ? failure : Stream.concat(Stream.fromArray(prefix), failure); + }), ); const client = { [ORCHESTRATION_WS_METHODS.subscribeThread]: (input: { @@ -244,7 +253,10 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o ); yield* SubscriptionRef.changes(threadState).pipe( Stream.runForEach((state) => - Ref.set(latest, state).pipe(Effect.andThen(Queue.offer(observed, state))), + Ref.update(stateChangeCount, (count) => count + 1).pipe( + Effect.andThen(Ref.set(latest, state)), + Effect.andThen(Queue.offer(observed, state)), + ), ), Effect.forkScoped, ); @@ -254,6 +266,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o inputs, observed, latest, + stateChangeCount, retryCount, subscriptionCount, loaderCalls, @@ -307,6 +320,38 @@ const titleUpdated = (title: string, sequence = 2): OrchestrationThreadStreamIte }, }); +const sessionSet = ( + status: "ready" | "running", + turnId: string, + sequence: number, +): OrchestrationThreadStreamItem => ({ + kind: "event", + event: { + eventId: EventId.make(`event-session-${status}-${sequence}`), + sequence, + occurredAt: "2026-04-01T03:00:00.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + aggregateKind: "thread", + aggregateId: THREAD_ID, + type: "thread.session-set", + payload: { + threadId: THREAD_ID, + session: { + threadId: THREAD_ID, + status, + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: status === "running" ? TurnId.make(turnId) : null, + lastError: null, + updatedAt: "2026-04-01T03:00:00.000Z", + }, + }, + }, +}); + const deleted = (): OrchestrationThreadStreamItem => ({ kind: "event", event: { @@ -989,4 +1034,36 @@ describe("EnvironmentThreads", () => { expect(yield* Ref.get(harness.subscriptionCount)).toBe(3); }), ); + + it.effect( + "persists a turn that settles mid-batch when the next turn starts in the same batch", + () => + Effect.gen(function* () { + const harness = yield* makeHarness({ cached: ACTIVE_THREAD }); + yield* awaitThreadState(harness.observed, (value) => value.status === "live"); + const before = yield* Ref.get(harness.stateChangeCount); + + // Both events arrive in one transport batch: the session settles and the + // next turn starts before the fold publishes. + yield* Queue.offerAll(harness.inputs, [ + sessionSet("ready", "turn-1", CACHED_SNAPSHOT_SEQUENCE + 1), + sessionSet("running", "turn-2", CACHED_SNAPSHOT_SEQUENCE + 2), + ]); + yield* awaitThreadState( + harness.observed, + (value) => + Option.isSome(value.data) && + value.data.value.session?.activeTurnId === TurnId.make("turn-2"), + ); + expect((yield* Ref.get(harness.stateChangeCount)) - before).toBe(1); + yield* TestClock.adjust("500 millis"); + yield* Effect.yieldNow; + + // The settled state reached the cache under its own sequence even + // though the batch ended on a running session. + const saved = (yield* Ref.get(harness.savedThreads)).at(-1); + expect(saved?.thread.session?.status).toBe("ready"); + expect(saved?.snapshotSequence).toBe(CACHED_SNAPSHOT_SEQUENCE + 1); + }), + ); }); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 0a44302e41ec51d1a17c47f61b19ac8944cf0c48..ab723cef485754a442c5784ded8ce38a1b731dc9 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -353,6 +353,29 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ), ); + const offerThreadPersistence = Effect.fn("EnvironmentThreadState.offerThreadPersistence")( + function* (thread: OrchestrationThread, snapshotSequence: number) { + const currentPage = yield* SubscriptionRef.get(state).pipe(Effect.map((value) => value.page)); + yield* Queue.offer(persistence, { + snapshotSequence, + thread, + // Persist the window boundary with the window's content so a cache + // restore can keep paging from where the loaded history ends. + ...Option.match(currentPage, { + onNone: () => ({}), + onSome: (value) => + ({ + page: { + beforeCursor: value.beforeCursor, + hasMore: value.hasMore, + snapshotSequence, + }, + }) as const, + }), + }); + }, + ); + const setThread = Effect.fn("EnvironmentThreadState.setThread")(function* ( thread: OrchestrationThread, // "keep" preserves the current page state (live events touch only loaded @@ -376,24 +399,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make // persist once it settles so cache encoding stays off the streaming path. if (shouldPersistThread(thread)) { const snapshotSequence = yield* SubscriptionRef.get(lastSequence); - const currentPage = yield* SubscriptionRef.get(state).pipe(Effect.map((value) => value.page)); - yield* Queue.offer(persistence, { - snapshotSequence, - thread, - // Persist the window boundary with the window's content so a cache - // restore can keep paging from where the loaded history ends. - ...Option.match(currentPage, { - onNone: () => ({}), - onSome: (value) => - ({ - page: { - beforeCursor: value.beforeCursor, - hasMore: value.hasMore, - snapshotSequence, - }, - }) as const, - }), - }); + yield* offerThreadPersistence(thread, snapshotSequence); } }); @@ -441,6 +447,9 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make // in the preserved history with no event left to remove it. The // epoch bump discards any older-page fetch racing this snapshot. yield* Ref.update(historyEpoch, (epoch) => epoch + 1); + // A parked response must not clear loadingOlder on a request started + // from the replacement snapshot's cursor. + yield* Ref.set(pendingOlderPage, null); yield* SubscriptionRef.set(lastSequence, item.snapshot.snapshotSequence); yield* setThread(item.snapshot.thread, pageStateFromSnapshot(item.snapshot.page)); return; @@ -539,17 +548,26 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make let thread = current.data.value; let sequence = yield* SubscriptionRef.get(lastSequence); let synchronized = false; + // Retain the last settled state even if the next turn starts before + // this batch publishes. Its cursor must describe that settled content. + let persistable: { thread: OrchestrationThread; sequence: number } | undefined; for (const item of items) { if (item.kind === "synchronized") { synchronized = true; } else if (item.kind === "event" && item.event.sequence > sequence) { sequence = item.event.sequence; const result = applyThreadDetailEvent(thread, item.event); - if (result.kind === "updated") thread = result.thread; + if (result.kind === "updated") { + thread = result.thread; + if (shouldPersistThread(thread)) persistable = { thread, sequence }; + } } } yield* SubscriptionRef.set(lastSequence, sequence); if (thread !== current.data.value) yield* setThread(thread, "keep"); + if (persistable !== undefined && !shouldPersistThread(thread)) { + yield* offerThreadPersistence(persistable.thread, persistable.sequence); + } if (synchronized) yield* applyItemLocked({ kind: "synchronized" }); yield* remember; }), -- 2.54.0