From ef80b8424df21d8358004dbba7d2eb1e5e8b32a1 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Sat, 10 Oct 2026 01:05:14 -0700 Subject: [PATCH 1/2] fix(client-runtime): queued runs that start after a steer show up in the thread A queued run keeps the run ordinal it got when it was queued, and the server bands its turn items by that ordinal. A steer or new turn sent later gets a higher ordinal and runs first, and on a partially loaded thread its items raise the client's latestLocalTurnOrdinal watermark. When the queued run finally started, every one of its items sat below the watermark, so the partial-timeline guard treated them as unloaded history and dropped them. The resume cursor still moved past those events and the thread cache kept the gap, so reloading did not bring them back. The guard now keeps a missing item when its run is still queued in the client's projection (the orchestrator commits a queued run's first item before the run.updated that starts it), and when the window already shows an earlier item of the same run (which also covers #16437's case). Items of other runs below the window are still dropped. Thread cache entries now carry their own schema version (4), so web and mobile discard thread snapshots saved with the gap and fetch a fresh bounded snapshot. Shell caches keep version 3 and stay warm. Fixes #16987 Co-Authored-By: Claude Opus 5.5 --- .../environment-cache-store.test.ts | 3 +- .../src/connection/environment-cache-store.ts | 11 +- apps/web/src/connection/storage.ts | 3 +- .../src/platform/orchestrationCache.test.ts | 22 +++- .../src/platform/orchestrationCache.ts | 7 +- .../state/orchestrationV2Projection.test.ts | 105 ++++++++++++++++++ .../src/state/orchestrationV2Projection.ts | 28 +++++ 7 files changed, 172 insertions(+), 7 deletions(-) diff --git a/apps/mobile/src/connection/environment-cache-store.test.ts b/apps/mobile/src/connection/environment-cache-store.test.ts index 5548a0fc7c88..4514c20e8564 100644 --- a/apps/mobile/src/connection/environment-cache-store.test.ts +++ b/apps/mobile/src/connection/environment-cache-store.test.ts @@ -1,5 +1,6 @@ import { ORCHESTRATION_CACHE_SCHEMA_VERSION, + ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, StoredOrchestrationShellSnapshot, } from "@t3tools/client-runtime/platform"; import { @@ -251,7 +252,7 @@ describe("mobile SQLite environment cache store", () => { ORCHESTRATION_CACHE_SCHEMA_VERSION, ); expect(memory.schemaVersions.get(cacheId(ENVIRONMENT_ID, "thread", THREAD_ID))).toBe( - ORCHESTRATION_CACHE_SCHEMA_VERSION, + ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, ); }), ); diff --git a/apps/mobile/src/connection/environment-cache-store.ts b/apps/mobile/src/connection/environment-cache-store.ts index d59b80dbc7e2..653aded1657d 100644 --- a/apps/mobile/src/connection/environment-cache-store.ts +++ b/apps/mobile/src/connection/environment-cache-store.ts @@ -1,5 +1,6 @@ import { ORCHESTRATION_CACHE_SCHEMA_VERSION, + ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, StoredOrchestrationThreadSnapshot, Persistence, } from "@t3tools/client-runtime/platform"; @@ -148,13 +149,19 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () { saveThread: Effect.fn("MobileEnvironmentCache.saveThread")(function* (environmentId, snapshot) { const threadId = snapshot.projection.thread.id; const payload = yield* encodeStoredThreadSnapshot({ - schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, + schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, environmentId, threadId, snapshot, }).pipe(Effect.mapError((cause) => persistenceError("save-thread", cause))); yield* database - .saveCache(environmentId, "thread", threadId, ORCHESTRATION_CACHE_SCHEMA_VERSION, payload) + .saveCache( + environmentId, + "thread", + threadId, + ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, + payload, + ) .pipe(Effect.mapError(mapDatabaseError("save-thread"))); }), removeThread: Effect.fn("MobileEnvironmentCache.removeThread")((environmentId, threadId) => diff --git a/apps/web/src/connection/storage.ts b/apps/web/src/connection/storage.ts index 15fca39e174e..8be169fccd45 100644 --- a/apps/web/src/connection/storage.ts +++ b/apps/web/src/connection/storage.ts @@ -3,6 +3,7 @@ import { type ConnectionCatalogDocument as ConnectionCatalogDocumentType, EMPTY_CONNECTION_CATALOG_DOCUMENT, ORCHESTRATION_CACHE_SCHEMA_VERSION, + ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, StoredOrchestrationShellSnapshot, StoredOrchestrationThreadSnapshot, decodeOrDiscardOrchestrationCache, @@ -810,7 +811,7 @@ export const layer = Layer.effectContext( saveThread: (environmentId, snapshot) => Effect.gen(function* () { const encoded = yield* encodeStoredThreadSnapshot({ - schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, + schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, environmentId, threadId: snapshot.projection.thread.id, snapshot, diff --git a/packages/client-runtime/src/platform/orchestrationCache.test.ts b/packages/client-runtime/src/platform/orchestrationCache.test.ts index 39170fe8cb81..06f4987c557b 100644 --- a/packages/client-runtime/src/platform/orchestrationCache.test.ts +++ b/packages/client-runtime/src/platform/orchestrationCache.test.ts @@ -15,6 +15,7 @@ import * as Schema from "effect/Schema"; import { ORCHESTRATION_CACHE_SCHEMA_VERSION, + ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, StoredOrchestrationShellSnapshot, StoredOrchestrationThreadSnapshot, decodeOrDiscardOrchestrationCache, @@ -102,7 +103,7 @@ describe("orchestration cache envelopes", () => { expect(projection).not.toBeNull(); if (projection === null) throw new Error("Expected live notice projection"); const encoded = encodeStoredThreadSnapshotJson({ - schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, + schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, environmentId, threadId: v2ThreadId, snapshot: { snapshotSequence: 5, projection }, @@ -121,7 +122,7 @@ describe("orchestration cache envelopes", () => { }); const shell = decodeStoredShellSnapshotJson(encodedShell); const encodedThread = encodeStoredThreadSnapshotJson({ - schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, + schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, environmentId, threadId: v2ThreadId, snapshot: { snapshotSequence: 4, projection: v2Projection }, @@ -159,6 +160,23 @@ describe("orchestration cache envelopes", () => { }), ); + it("rejects thread snapshots saved before queued-run items were kept, but not shells", () => { + // Version 3 thread windows may have dropped a queued run's items (#16987). + const staleThread = encodeStoredThreadSnapshotJson({ + schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, + environmentId, + threadId: v2ThreadId, + snapshot: { snapshotSequence: 4, projection: v2Projection }, + }).replace(`"schemaVersion":${ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION}`, `"schemaVersion":3`); + expect(() => decodeStoredThreadSnapshotJson(staleThread)).toThrow(); + const shell = encodeStoredShellSnapshot({ + schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, + environmentId, + snapshot: v2ShellSnapshot, + }); + expect(decodeStoredShellSnapshotSync(shell).schemaVersion).toBe(3); + }); + it.effect("discards V1-versioned cache envelopes after a decode failure", () => Effect.gen(function* () { let discardCount = 0; diff --git a/packages/client-runtime/src/platform/orchestrationCache.ts b/packages/client-runtime/src/platform/orchestrationCache.ts index e39fe6c1dfe6..4f1f910827f1 100644 --- a/packages/client-runtime/src/platform/orchestrationCache.ts +++ b/packages/client-runtime/src/platform/orchestrationCache.ts @@ -10,6 +10,11 @@ import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; export const ORCHESTRATION_CACHE_SCHEMA_VERSION = 3 as const; +/** + * Thread snapshots version on their own so a thread-only invalidation keeps + * shell caches warm. 4 drops bounded windows that lost queued runs' items (#16987). + */ +export const ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION = 4 as const; export const StoredOrchestrationShellSnapshot = Schema.Struct({ schemaVersion: Schema.Literal(ORCHESTRATION_CACHE_SCHEMA_VERSION), @@ -18,7 +23,7 @@ export const StoredOrchestrationShellSnapshot = Schema.Struct({ }); export const StoredOrchestrationThreadSnapshot = Schema.Struct({ - schemaVersion: Schema.Literal(ORCHESTRATION_CACHE_SCHEMA_VERSION), + schemaVersion: Schema.Literal(ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION), environmentId: EnvironmentId, threadId: ThreadId, snapshot: OrchestrationV2ThreadDetailSnapshot.mapFields((fields) => ({ diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts index 1a53afb8d7e8..16d78e876a2d 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts @@ -275,6 +275,111 @@ describe("applyOrchestrationV2ProjectionEvent", () => { expect(next?.visibleTurnItems.map((row) => row.position)).toEqual([0, 1]); }); + describe("partial timeline", () => { + const queuedRunId = RunId.make("run-queued"); + const steerRunId = RunId.make("run-steer"); + const queuedRun = { ...run, id: queuedRunId, ordinal: 2, status: "queued" } as const; + const steerRun = { ...run, id: steerRunId, ordinal: 3, status: "completed" } as const; + const runItem = (id: string, itemRunId: RunId, ordinal: number) => ({ + ...commandItem(id, id, ordinal), + runId: itemRunId, + }); + const windowOf = ( + runs: ReadonlyArray, + items: ReadonlyArray, + ): OrchestrationV2ThreadProjection => ({ + ...emptyProjection, + runs, + turnItems: items, + visibleTurnItems: items.map((item, position) => ({ + position, + visibility: "local" as const, + sourceThreadId: threadId, + sourceItemId: item.id, + item, + })), + }); + const itemEvent = (payload: OrchestrationV2TurnItem) => + ({ + id: `event-${payload.id}`, + type: "turn-item.updated", + threadId, + occurredAt: now, + payload, + }) as OrchestrationV2DomainEvent; + const runEvent = (payload: OrchestrationV2Run) => + ({ + id: `event-${payload.id}-${payload.status}`, + type: "run.updated", + threadId, + occurredAt: now, + payload, + }) as OrchestrationV2DomainEvent; + + it("shows a queued run that starts after a newer steer raised the watermark", () => { + // Run 2 was queued first, run 3 was steered in and ran first. Its items + // sit in run 3's band and already raised the watermark past run 2's band. + const steerItems = [ + runItem("item-steer-user", steerRunId, 3_000_001), + runItem("item-steer-reply", steerRunId, 3_000_002), + ]; + let projection: OrchestrationV2ThreadProjection | null = windowOf( + [run, queuedRun, steerRun], + steerItems, + ); + let latestLocalTurnOrdinal = 3_000_002; + // Same order the orchestrator commits a queued start: the user message + // lands while the run is still queued, then the run starts and replies. + const queuedUser = runItem("item-queued-user", queuedRunId, 2_000_001); + const queuedReply = runItem("item-queued-reply", queuedRunId, 2_000_002); + for (const event of [ + itemEvent(queuedUser), + runEvent({ ...queuedRun, status: "running" }), + itemEvent(queuedReply), + runEvent({ ...queuedRun, status: "completed" }), + ]) { + projection = applyOrchestrationV2ProjectionEvent(projection, event, { + partialTimeline: true, + latestLocalTurnOrdinal, + }); + if (event.type === "turn-item.updated") { + latestLocalTurnOrdinal = Math.max(latestLocalTurnOrdinal, event.payload.ordinal); + } + } + + expect(projection?.visibleTurnItems.map((row) => row.item.id)).toEqual([ + queuedUser.id, + queuedReply.id, + ...steerItems.map((item) => item.id), + ]); + expect(projection?.turnItems.map((item) => item.id)).toContain(queuedReply.id); + }); + + it.each([ + { name: "keeps an item of a run already in the window", itemRunId: steerRunId, kept: true }, + { name: "drops an item of a finished run outside the window", itemRunId: runId, kept: false }, + ])("$name that arrives below the watermark", ({ itemRunId, kept }) => { + const first = runItem("item-first", steerRunId, 3_000_001); + const later = runItem("item-later", steerRunId, 3_000_003); + const late = runItem("item-late", itemRunId, 3_000_002); + const projection = windowOf([run, steerRun], [first, later]); + + const next = applyOrchestrationV2ProjectionEvent(projection, itemEvent(late), { + partialTimeline: true, + latestLocalTurnOrdinal: later.ordinal, + }); + if (kept) { + expect(next?.visibleTurnItems.map((row) => row.item.id)).toEqual([ + first.id, + late.id, + later.id, + ]); + } else { + expect(next).toBe(projection); + } + }); + }); + it("removes only hidden local items while preserving inherited rows", () => { const inherited = commandItem("item-inherited"); const local = commandItem("item-local"); diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.ts b/packages/client-runtime/src/state/orchestrationV2Projection.ts index 7e96a3039353..6d94ccde7bfb 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.ts @@ -94,6 +94,34 @@ function shouldDropMissingPartialTurnItem( if (projection.visibleTurnItems.some((row) => row.sourceItemId === item.id)) { return false; } + if (item.runId === null) { + return isBelowPartialWindow(projection, item, latestLocalTurnOrdinal); + } + // A queued run keeps the ordinal it got when queued, so a steer sent later + // can raise the watermark past it. The orchestrator commits its first item + // while the run is still queued, so that item is new, not unloaded history. + if (projection.runs.some((run) => run.id === item.runId && run.status === "queued")) { + return false; + } + // Later items of a run already in the window can raise the watermark first. + if ( + projection.visibleTurnItems.some( + (row) => + row.visibility === "local" && + row.item.runId === item.runId && + row.item.ordinal <= item.ordinal, + ) + ) { + return false; + } + return isBelowPartialWindow(projection, item, latestLocalTurnOrdinal); +} + +function isBelowPartialWindow( + projection: OrchestrationV2ThreadProjection, + item: OrchestrationV2TurnItem, + latestLocalTurnOrdinal: number | null | undefined, +): boolean { if ( latestLocalTurnOrdinal !== null && latestLocalTurnOrdinal !== undefined && From d3f7855c36bcd76fce441c37b502fbaafed1343e Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Sat, 10 Oct 2026 01:52:35 -0700 Subject: [PATCH 2/2] fix(client-runtime): a running run's items show even when the window starts after its earlier ones A bounded snapshot keeps every run but only the recent window's items. A run that started before the window and is still running had no row in it, so its next item fell under both keep rules and was dropped. Keep a missing item whose run has not finished (preparing, queued, starting, running or waiting), not only a queued one. Co-Authored-By: Claude Opus 5.5 --- .../state/orchestrationV2Projection.test.ts | 22 +++++++++++++++++++ .../src/state/orchestrationV2Projection.ts | 18 ++++++++++++--- 2 files changed, 37 insertions(+), 3 deletions(-) diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts index 16d78e876a2d..e41eb81257f9 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts @@ -355,6 +355,28 @@ describe("applyOrchestrationV2ProjectionEvent", () => { expect(projection?.turnItems.map((item) => item.id)).toContain(queuedReply.id); }); + it("keeps a running run's item when the window starts after its earlier items", () => { + // The bounded window begins inside run 3; run 2 started and kept going, + // but none of its rows made it into the window. + const runningRun = { ...queuedRun, status: "running" } as const; + const steerItems = [ + runItem("item-steer-user", steerRunId, 3_000_001), + runItem("item-steer-reply", steerRunId, 3_000_002), + ]; + const projection = windowOf([run, runningRun, steerRun], steerItems); + const reply = runItem("item-running-reply", queuedRunId, 2_000_003); + + const next = applyOrchestrationV2ProjectionEvent(projection, itemEvent(reply), { + partialTimeline: true, + latestLocalTurnOrdinal: 3_000_002, + }); + + expect(next?.visibleTurnItems.map((row) => row.item.id)).toEqual([ + reply.id, + ...steerItems.map((item) => item.id), + ]); + }); + it.each([ { name: "keeps an item of a run already in the window", itemRunId: steerRunId, kept: true }, { name: "drops an item of a finished run outside the window", itemRunId: runId, kept: false }, diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.ts b/packages/client-runtime/src/state/orchestrationV2Projection.ts index 6d94ccde7bfb..4587221c07df 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.ts @@ -1,5 +1,6 @@ import type { OrchestrationV2DomainEvent, + OrchestrationV2Run, OrchestrationV2ThreadProjection, OrchestrationV2TurnItem, } from "@t3tools/contracts"; @@ -86,6 +87,14 @@ function oldestLocalTurnOrdinal( return oldest; } +const UNFINISHED_RUN_STATUSES: ReadonlySet = new Set([ + "preparing", + "queued", + "starting", + "running", + "waiting", +]); + function shouldDropMissingPartialTurnItem( projection: OrchestrationV2ThreadProjection, item: OrchestrationV2TurnItem, @@ -98,9 +107,12 @@ function shouldDropMissingPartialTurnItem( return isBelowPartialWindow(projection, item, latestLocalTurnOrdinal); } // A queued run keeps the ordinal it got when queued, so a steer sent later - // can raise the watermark past it. The orchestrator commits its first item - // while the run is still queued, so that item is new, not unloaded history. - if (projection.runs.some((run) => run.id === item.runId && run.status === "queued")) { + // can raise the watermark past it. Items of a run that has not finished are + // new, not unloaded history: the first lands while the run is still queued, + // and a bounded window can start after the ones before it. + if ( + projection.runs.some((run) => run.id === item.runId && UNFINISHED_RUN_STATUSES.has(run.status)) + ) { return false; } // Later items of a run already in the window can raise the watermark first.