authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-10-10 12:31:57-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-10 12:31:57-07:00
log61eb4af69c38c1f39e58ed76aa695b751f9f9baf
tree5a213ffa9c9a9d0469d8a6cf80b84f14183220df
parentec4378614046ab58864ff5e640de2b46f6704505
signature Signed by PGP key B5690EEEBB952194

fix(client-runtime): queued runs that start after a steer show up in the thread (#17764)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

7 files changed, 206 insertions(+), 7 deletions(-)

apps/mobile/src/connection/environment-cache-store.test.ts+2-1
...@@ -1,5 +1,6 @@...@@ -1,5 +1,6 @@
1import {1import {
2 ORCHESTRATION_CACHE_SCHEMA_VERSION,2 ORCHESTRATION_CACHE_SCHEMA_VERSION,
3 ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
3 StoredOrchestrationShellSnapshot,4 StoredOrchestrationShellSnapshot,
4} from "@t3tools/client-runtime/platform";5} from "@t3tools/client-runtime/platform";
5import {6import {
...@@ -251,7 +252,7 @@ describe("mobile SQLite environment cache store", () => {...@@ -251,7 +252,7 @@ describe("mobile SQLite environment cache store", () => {
251 ORCHESTRATION_CACHE_SCHEMA_VERSION,252 ORCHESTRATION_CACHE_SCHEMA_VERSION,
252 );253 );
253 expect(memory.schemaVersions.get(cacheId(ENVIRONMENT_ID, "thread", THREAD_ID))).toBe(254 expect(memory.schemaVersions.get(cacheId(ENVIRONMENT_ID, "thread", THREAD_ID))).toBe(
254 ORCHESTRATION_CACHE_SCHEMA_VERSION,255 ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
255 );256 );
256 }),257 }),
257 );258 );
apps/mobile/src/connection/environment-cache-store.ts+9-2
...@@ -1,5 +1,6 @@...@@ -1,5 +1,6 @@
1import {1import {
2 ORCHESTRATION_CACHE_SCHEMA_VERSION,2 ORCHESTRATION_CACHE_SCHEMA_VERSION,
3 ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
3 StoredOrchestrationThreadSnapshot,4 StoredOrchestrationThreadSnapshot,
4 Persistence,5 Persistence,
5} from "@t3tools/client-runtime/platform";6} from "@t3tools/client-runtime/platform";
...@@ -148,13 +149,19 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () {...@@ -148,13 +149,19 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () {
148 saveThread: Effect.fn("MobileEnvironmentCache.saveThread")(function* (environmentId, snapshot) {149 saveThread: Effect.fn("MobileEnvironmentCache.saveThread")(function* (environmentId, snapshot) {
149 const threadId = snapshot.projection.thread.id;150 const threadId = snapshot.projection.thread.id;
150 const payload = yield* encodeStoredThreadSnapshot({151 const payload = yield* encodeStoredThreadSnapshot({
151 schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION,152 schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
152 environmentId,153 environmentId,
153 threadId,154 threadId,
154 snapshot,155 snapshot,
155 }).pipe(Effect.mapError((cause) => persistenceError("save-thread", cause)));156 }).pipe(Effect.mapError((cause) => persistenceError("save-thread", cause)));
156 yield* database157 yield* database
157 .saveCache(environmentId, "thread", threadId, ORCHESTRATION_CACHE_SCHEMA_VERSION, payload)158 .saveCache(
159 environmentId,
160 "thread",
161 threadId,
162 ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
163 payload,
164 )
158 .pipe(Effect.mapError(mapDatabaseError("save-thread")));165 .pipe(Effect.mapError(mapDatabaseError("save-thread")));
159 }),166 }),
160 removeThread: Effect.fn("MobileEnvironmentCache.removeThread")((environmentId, threadId) =>167 removeThread: Effect.fn("MobileEnvironmentCache.removeThread")((environmentId, threadId) =>
apps/web/src/connection/storage.ts+2-1
...@@ -3,6 +3,7 @@ import {...@@ -3,6 +3,7 @@ import {
3 type ConnectionCatalogDocument as ConnectionCatalogDocumentType,3 type ConnectionCatalogDocument as ConnectionCatalogDocumentType,
4 EMPTY_CONNECTION_CATALOG_DOCUMENT,4 EMPTY_CONNECTION_CATALOG_DOCUMENT,
5 ORCHESTRATION_CACHE_SCHEMA_VERSION,5 ORCHESTRATION_CACHE_SCHEMA_VERSION,
6 ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
6 StoredOrchestrationShellSnapshot,7 StoredOrchestrationShellSnapshot,
7 StoredOrchestrationThreadSnapshot,8 StoredOrchestrationThreadSnapshot,
8 decodeOrDiscardOrchestrationCache,9 decodeOrDiscardOrchestrationCache,
...@@ -810,7 +811,7 @@ export const layer = Layer.effectContext(...@@ -810,7 +811,7 @@ export const layer = Layer.effectContext(
810 saveThread: (environmentId, snapshot) =>811 saveThread: (environmentId, snapshot) =>
811 Effect.gen(function* () {812 Effect.gen(function* () {
812 const encoded = yield* encodeStoredThreadSnapshot({813 const encoded = yield* encodeStoredThreadSnapshot({
813 schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION,814 schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
814 environmentId,815 environmentId,
815 threadId: snapshot.projection.thread.id,816 threadId: snapshot.projection.thread.id,
816 snapshot,817 snapshot,
packages/client-runtime/src/platform/orchestrationCache.test.ts+20-2
...@@ -15,6 +15,7 @@ import * as Schema from "effect/Schema";...@@ -15,6 +15,7 @@ import * as Schema from "effect/Schema";
1515
16import {16import {
17 ORCHESTRATION_CACHE_SCHEMA_VERSION,17 ORCHESTRATION_CACHE_SCHEMA_VERSION,
18 ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
18 StoredOrchestrationShellSnapshot,19 StoredOrchestrationShellSnapshot,
19 StoredOrchestrationThreadSnapshot,20 StoredOrchestrationThreadSnapshot,
20 decodeOrDiscardOrchestrationCache,21 decodeOrDiscardOrchestrationCache,
...@@ -102,7 +103,7 @@ describe("orchestration cache envelopes", () => {...@@ -102,7 +103,7 @@ describe("orchestration cache envelopes", () => {
102 expect(projection).not.toBeNull();103 expect(projection).not.toBeNull();
103 if (projection === null) throw new Error("Expected live notice projection");104 if (projection === null) throw new Error("Expected live notice projection");
104 const encoded = encodeStoredThreadSnapshotJson({105 const encoded = encodeStoredThreadSnapshotJson({
105 schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION,106 schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
106 environmentId,107 environmentId,
107 threadId: v2ThreadId,108 threadId: v2ThreadId,
108 snapshot: { snapshotSequence: 5, projection },109 snapshot: { snapshotSequence: 5, projection },
...@@ -121,7 +122,7 @@ describe("orchestration cache envelopes", () => {...@@ -121,7 +122,7 @@ describe("orchestration cache envelopes", () => {
121 });122 });
122 const shell = decodeStoredShellSnapshotJson(encodedShell);123 const shell = decodeStoredShellSnapshotJson(encodedShell);
123 const encodedThread = encodeStoredThreadSnapshotJson({124 const encodedThread = encodeStoredThreadSnapshotJson({
124 schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION,125 schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
125 environmentId,126 environmentId,
126 threadId: v2ThreadId,127 threadId: v2ThreadId,
127 snapshot: { snapshotSequence: 4, projection: v2Projection },128 snapshot: { snapshotSequence: 4, projection: v2Projection },
...@@ -159,6 +160,23 @@ describe("orchestration cache envelopes", () => {...@@ -159,6 +160,23 @@ describe("orchestration cache envelopes", () => {
159 }),160 }),
160 );161 );
161162
163 it("rejects thread snapshots saved before queued-run items were kept, but not shells", () => {
164 // Version 3 thread windows may have dropped a queued run's items (#16987).
165 const staleThread = encodeStoredThreadSnapshotJson({
166 schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
167 environmentId,
168 threadId: v2ThreadId,
169 snapshot: { snapshotSequence: 4, projection: v2Projection },
170 }).replace(`"schemaVersion":${ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION}`, `"schemaVersion":3`);
171 expect(() => decodeStoredThreadSnapshotJson(staleThread)).toThrow();
172 const shell = encodeStoredShellSnapshot({
173 schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION,
174 environmentId,
175 snapshot: v2ShellSnapshot,
176 });
177 expect(decodeStoredShellSnapshotSync(shell).schemaVersion).toBe(3);
178 });
179
162 it.effect("discards V1-versioned cache envelopes after a decode failure", () =>180 it.effect("discards V1-versioned cache envelopes after a decode failure", () =>
163 Effect.gen(function* () {181 Effect.gen(function* () {
164 let discardCount = 0;182 let discardCount = 0;
packages/client-runtime/src/platform/orchestrationCache.ts+6-1
...@@ -10,6 +10,11 @@ import * as Option from "effect/Option";...@@ -10,6 +10,11 @@ import * as Option from "effect/Option";
10import * as Schema from "effect/Schema";10import * as Schema from "effect/Schema";
1111
12export const ORCHESTRATION_CACHE_SCHEMA_VERSION = 3 as const;12export const ORCHESTRATION_CACHE_SCHEMA_VERSION = 3 as const;
13/**
14 * Thread snapshots version on their own so a thread-only invalidation keeps
15 * shell caches warm. 4 drops bounded windows that lost queued runs' items (#16987).
16 */
17export const ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION = 4 as const;
1318
14export const StoredOrchestrationShellSnapshot = Schema.Struct({19export const StoredOrchestrationShellSnapshot = Schema.Struct({
15 schemaVersion: Schema.Literal(ORCHESTRATION_CACHE_SCHEMA_VERSION),20 schemaVersion: Schema.Literal(ORCHESTRATION_CACHE_SCHEMA_VERSION),
...@@ -18,7 +23,7 @@ export const StoredOrchestrationShellSnapshot = Schema.Struct({...@@ -18,7 +23,7 @@ export const StoredOrchestrationShellSnapshot = Schema.Struct({
18});23});
1924
20export const StoredOrchestrationThreadSnapshot = Schema.Struct({25export const StoredOrchestrationThreadSnapshot = Schema.Struct({
21 schemaVersion: Schema.Literal(ORCHESTRATION_CACHE_SCHEMA_VERSION),26 schemaVersion: Schema.Literal(ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION),
22 environmentId: EnvironmentId,27 environmentId: EnvironmentId,
23 threadId: ThreadId,28 threadId: ThreadId,
24 snapshot: OrchestrationV2ThreadDetailSnapshot.mapFields((fields) => ({29 snapshot: OrchestrationV2ThreadDetailSnapshot.mapFields((fields) => ({
packages/client-runtime/src/state/orchestrationV2Projection.test.ts+127
...@@ -275,6 +275,133 @@ describe("applyOrchestrationV2ProjectionEvent", () => {...@@ -275,6 +275,133 @@ describe("applyOrchestrationV2ProjectionEvent", () => {
275 expect(next?.visibleTurnItems.map((row) => row.position)).toEqual([0, 1]);275 expect(next?.visibleTurnItems.map((row) => row.position)).toEqual([0, 1]);
276 });276 });
277277
278 describe("partial timeline", () => {
279 const queuedRunId = RunId.make("run-queued");
280 const steerRunId = RunId.make("run-steer");
281 const queuedRun = { ...run, id: queuedRunId, ordinal: 2, status: "queued" } as const;
282 const steerRun = { ...run, id: steerRunId, ordinal: 3, status: "completed" } as const;
283 const runItem = (id: string, itemRunId: RunId, ordinal: number) => ({
284 ...commandItem(id, id, ordinal),
285 runId: itemRunId,
286 });
287 const windowOf = (
288 runs: ReadonlyArray<OrchestrationV2Run>,
289 items: ReadonlyArray<OrchestrationV2TurnItem>,
290 ): OrchestrationV2ThreadProjection => ({
291 ...emptyProjection,
292 runs,
293 turnItems: items,
294 visibleTurnItems: items.map((item, position) => ({
295 position,
296 visibility: "local" as const,
297 sourceThreadId: threadId,
298 sourceItemId: item.id,
299 item,
300 })),
301 });
302 const itemEvent = (payload: OrchestrationV2TurnItem) =>
303 ({
304 id: `event-${payload.id}`,
305 type: "turn-item.updated",
306 threadId,
307 occurredAt: now,
308 payload,
309 }) as OrchestrationV2DomainEvent;
310 const runEvent = (payload: OrchestrationV2Run) =>
311 ({
312 id: `event-${payload.id}-${payload.status}`,
313 type: "run.updated",
314 threadId,
315 occurredAt: now,
316 payload,
317 }) as OrchestrationV2DomainEvent;
318
319 it("shows a queued run that starts after a newer steer raised the watermark", () => {
320 // Run 2 was queued first, run 3 was steered in and ran first. Its items
321 // sit in run 3's band and already raised the watermark past run 2's band.
322 const steerItems = [
323 runItem("item-steer-user", steerRunId, 3_000_001),
324 runItem("item-steer-reply", steerRunId, 3_000_002),
325 ];
326 let projection: OrchestrationV2ThreadProjection | null = windowOf(
327 [run, queuedRun, steerRun],
328 steerItems,
329 );
330 let latestLocalTurnOrdinal = 3_000_002;
331 // Same order the orchestrator commits a queued start: the user message
332 // lands while the run is still queued, then the run starts and replies.
333 const queuedUser = runItem("item-queued-user", queuedRunId, 2_000_001);
334 const queuedReply = runItem("item-queued-reply", queuedRunId, 2_000_002);
335 for (const event of [
336 itemEvent(queuedUser),
337 runEvent({ ...queuedRun, status: "running" }),
338 itemEvent(queuedReply),
339 runEvent({ ...queuedRun, status: "completed" }),
340 ]) {
341 projection = applyOrchestrationV2ProjectionEvent(projection, event, {
342 partialTimeline: true,
343 latestLocalTurnOrdinal,
344 });
345 if (event.type === "turn-item.updated") {
346 latestLocalTurnOrdinal = Math.max(latestLocalTurnOrdinal, event.payload.ordinal);
347 }
348 }
349
350 expect(projection?.visibleTurnItems.map((row) => row.item.id)).toEqual([
351 queuedUser.id,
352 queuedReply.id,
353 ...steerItems.map((item) => item.id),
354 ]);
355 expect(projection?.turnItems.map((item) => item.id)).toContain(queuedReply.id);
356 });
357
358 it("keeps a running run's item when the window starts after its earlier items", () => {
359 // The bounded window begins inside run 3; run 2 started and kept going,
360 // but none of its rows made it into the window.
361 const runningRun = { ...queuedRun, status: "running" } as const;
362 const steerItems = [
363 runItem("item-steer-user", steerRunId, 3_000_001),
364 runItem("item-steer-reply", steerRunId, 3_000_002),
365 ];
366 const projection = windowOf([run, runningRun, steerRun], steerItems);
367 const reply = runItem("item-running-reply", queuedRunId, 2_000_003);
368
369 const next = applyOrchestrationV2ProjectionEvent(projection, itemEvent(reply), {
370 partialTimeline: true,
371 latestLocalTurnOrdinal: 3_000_002,
372 });
373
374 expect(next?.visibleTurnItems.map((row) => row.item.id)).toEqual([
375 reply.id,
376 ...steerItems.map((item) => item.id),
377 ]);
378 });
379
380 it.each([
381 { name: "keeps an item of a run already in the window", itemRunId: steerRunId, kept: true },
382 { name: "drops an item of a finished run outside the window", itemRunId: runId, kept: false },
383 ])("$name that arrives below the watermark", ({ itemRunId, kept }) => {
384 const first = runItem("item-first", steerRunId, 3_000_001);
385 const later = runItem("item-later", steerRunId, 3_000_003);
386 const late = runItem("item-late", itemRunId, 3_000_002);
387 const projection = windowOf([run, steerRun], [first, later]);
388
389 const next = applyOrchestrationV2ProjectionEvent(projection, itemEvent(late), {
390 partialTimeline: true,
391 latestLocalTurnOrdinal: later.ordinal,
392 });
393 if (kept) {
394 expect(next?.visibleTurnItems.map((row) => row.item.id)).toEqual([
395 first.id,
396 late.id,
397 later.id,
398 ]);
399 } else {
400 expect(next).toBe(projection);
401 }
402 });
403 });
404
278 it("removes only hidden local items while preserving inherited rows", () => {405 it("removes only hidden local items while preserving inherited rows", () => {
279 const inherited = commandItem("item-inherited");406 const inherited = commandItem("item-inherited");
280 const local = commandItem("item-local");407 const local = commandItem("item-local");
packages/client-runtime/src/state/orchestrationV2Projection.ts+40
...@@ -1,5 +1,6 @@...@@ -1,5 +1,6 @@
1import type {1import type {
2 OrchestrationV2DomainEvent,2 OrchestrationV2DomainEvent,
3 OrchestrationV2Run,
3 OrchestrationV2ThreadProjection,4 OrchestrationV2ThreadProjection,
4 OrchestrationV2TurnItem,5 OrchestrationV2TurnItem,
5} from "@t3tools/contracts";6} from "@t3tools/contracts";
...@@ -86,6 +87,14 @@ function oldestLocalTurnOrdinal(...@@ -86,6 +87,14 @@ function oldestLocalTurnOrdinal(
86 return oldest;87 return oldest;
87}88}
8889
90const UNFINISHED_RUN_STATUSES: ReadonlySet<OrchestrationV2Run["status"]> = new Set([
91 "preparing",
92 "queued",
93 "starting",
94 "running",
95 "waiting",
96]);
97
89function shouldDropMissingPartialTurnItem(98function shouldDropMissingPartialTurnItem(
90 projection: OrchestrationV2ThreadProjection,99 projection: OrchestrationV2ThreadProjection,
91 item: OrchestrationV2TurnItem,100 item: OrchestrationV2TurnItem,
...@@ -94,6 +103,37 @@ function shouldDropMissingPartialTurnItem(...@@ -94,6 +103,37 @@ function shouldDropMissingPartialTurnItem(
94 if (projection.visibleTurnItems.some((row) => row.sourceItemId === item.id)) {103 if (projection.visibleTurnItems.some((row) => row.sourceItemId === item.id)) {
95 return false;104 return false;
96 }105 }
106 if (item.runId === null) {
107 return isBelowPartialWindow(projection, item, latestLocalTurnOrdinal);
108 }
109 // A queued run keeps the ordinal it got when queued, so a steer sent later
110 // can raise the watermark past it. Items of a run that has not finished are
111 // new, not unloaded history: the first lands while the run is still queued,
112 // and a bounded window can start after the ones before it.
113 if (
114 projection.runs.some((run) => run.id === item.runId && UNFINISHED_RUN_STATUSES.has(run.status))
115 ) {
116 return false;
117 }
118 // Later items of a run already in the window can raise the watermark first.
119 if (
120 projection.visibleTurnItems.some(
121 (row) =>
122 row.visibility === "local" &&
123 row.item.runId === item.runId &&
124 row.item.ordinal <= item.ordinal,
125 )
126 ) {
127 return false;
128 }
129 return isBelowPartialWindow(projection, item, latestLocalTurnOrdinal);
130}
131
132function isBelowPartialWindow(
133 projection: OrchestrationV2ThreadProjection,
134 item: OrchestrationV2TurnItem,
135 latestLocalTurnOrdinal: number | null | undefined,
136): boolean {
97 if (137 if (
98 latestLocalTurnOrdinal !== null &&138 latestLocalTurnOrdinal !== null &&
99 latestLocalTurnOrdinal !== undefined &&139 latestLocalTurnOrdinal !== undefined &&