From 41f339d37e3db50a3a8680740031b9d88dc0e0db Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 18 Oct 2026 23:00:46 -0500 Subject: [PATCH 1/6] perf(contracts): filter unknown timeline members at read boundaries Forward-compatible turn item arrays decoded each element through ForwardCompatibleUnion, dropped unknown members, then decoded to Schema.Array(Schema.toType(...)), which validated and rebuilt every known item a second time. Thread projections, detail snapshots and history pages paid it on every open and page. The decoded types now keep UnknownUnionMember (Received* types), so each known item is validated once. The exported projection, detail snapshot and history page types still hold only known items, and clients drop unknown ones where they read: socket and bounded snapshots, the full HTTP snapshot, history pages, the thread cache and getThreadProjection. Malformed known items still fail to decode. Refs #18856, #9661 Co-Authored-By: Claude Opus 5.5 --- apps/mobile/scripts/measure-renders.mts | 16 ++- apps/server/scripts/verify-background-live.ts | 6 +- .../src/orchestration-v2/ThreadStream.ts | 9 +- .../boundedSnapshotTransport.test.ts | 3 +- .../client-runtime/src/operations/commands.ts | 5 +- .../src/platform/orchestrationCache.test.ts | 5 +- .../src/platform/persistence.ts | 3 +- .../src/state/threadHistoryHttp.ts | 4 +- .../src/state/threadSnapshotHttp.ts | 8 +- packages/client-runtime/src/state/threads.ts | 2 + packages/contracts/src/baseSchemas.test.ts | 27 ++++- packages/contracts/src/baseSchemas.ts | 29 ++--- .../contracts/src/orchestrationV2.test.ts | 30 +++-- packages/contracts/src/orchestrationV2.ts | 112 ++++++++++++------ .../orchestrationV2BoundedSnapshot.test.ts | 25 ++++ .../src/orchestrationV2BoundedSnapshot.ts | 16 ++- 16 files changed, 211 insertions(+), 89 deletions(-) diff --git a/apps/mobile/scripts/measure-renders.mts b/apps/mobile/scripts/measure-renders.mts index 7ca1a8b82d737..0df2f84c5ca2c 100644 --- a/apps/mobile/scripts/measure-renders.mts +++ b/apps/mobile/scripts/measure-renders.mts @@ -15,7 +15,9 @@ import { ProviderInstanceId, ThreadId, WsRpcGroup, + knownThreadProjection, type OrchestrationV2ThreadProjection, + type ReceivedOrchestrationV2ThreadProjection, } from "@t3tools/contracts"; import * as Console from "effect/Console"; import * as Crypto from "effect/Crypto"; @@ -228,10 +230,14 @@ const measureStreamReply = Effect.fn("measure.streamReply")(function* ( yield* evaluate( `__t3RenderCounterOpen(${encodeJson(`${options.scheme}://threads/${environmentId}/${threadId}`)})`, ); - const projectionOf = (item: { kind: string; projection?: OrchestrationV2ThreadProjection }) => - item.kind === "snapshot" && item.projection + const projectionOf = (item: { + kind: string; + projection?: ReceivedOrchestrationV2ThreadProjection; + }) => + (item.kind === "snapshot" && item.projection ? Effect.succeed(item.projection) - : client["orchestration.getThreadProjection"]({ threadId }); + : client["orchestration.getThreadProjection"]({ threadId }) + ).pipe(Effect.map(knownThreadProjection)); const streaming = yield* client["orchestration.subscribeThread"]({ threadId }).pipe( Stream.mapEffect(projectionOf), Stream.filter((projection) => streamingItem(projection) !== undefined), @@ -249,7 +255,9 @@ const measureStreamReply = Effect.fn("measure.streamReply")(function* ( return yield* new MeasureError({ message: `Run ${run}: no React commits in the window. Check the device for a red box.`, }); - const after = yield* client["orchestration.getThreadProjection"]({ threadId }); + const after = knownThreadProjection( + yield* client["orchestration.getThreadProjection"]({ threadId }), + ); return { report, stillStreaming: streamingItem(after) !== undefined }; }, Effect.scoped); diff --git a/apps/server/scripts/verify-background-live.ts b/apps/server/scripts/verify-background-live.ts index 0279afe982b27..cde8b81a2be34 100644 --- a/apps/server/scripts/verify-background-live.ts +++ b/apps/server/scripts/verify-background-live.ts @@ -18,6 +18,7 @@ import { ORCHESTRATION_PROTOCOL_VERSION, ORCHESTRATION_PROTOCOL_HEADER, CommandId, + knownThreadProjection, ProjectId, ProviderInstanceId, ThreadId, @@ -374,10 +375,11 @@ try { NodePath.join(evidence, "events.ndjson"), JSON.stringify(item) + "\n", ); - const projection = + const projection = knownThreadProjection( item.kind === "snapshot" ? item.projection - : yield* client["orchestration.getThreadProjection"]({ threadId }); + : yield* client["orchestration.getThreadProjection"]({ threadId }), + ); const childThreadId = projection.subagents[0]?.childThreadId; if (values.scenario === "nested" && childThreadId && !watchingChild) { watchingChild = true; diff --git a/apps/server/src/orchestration-v2/ThreadStream.ts b/apps/server/src/orchestration-v2/ThreadStream.ts index 4ef21edd3b3c7..bf9596d161402 100644 --- a/apps/server/src/orchestration-v2/ThreadStream.ts +++ b/apps/server/src/orchestration-v2/ThreadStream.ts @@ -34,10 +34,11 @@ export function threadReplayEncodedBytes(items: ReadonlyArray): number return total; } -type ThreadSnapshotStreamItem = Extract< - OrchestrationV2ThreadStreamItem, - { readonly kind: "snapshot" } ->; +// Built from this server's own projection, so it holds only turn items it knows. +type ThreadSnapshotStreamItem = Omit< + Extract, + "projection" +> & { readonly projection: OrchestrationV2ThreadProjection }; export function boundedSnapshotResponseFields(input: { readonly bounded: BoundedProjectionResult; diff --git a/apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts b/apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts index 9519b661cbf3f..7a26e1e6ac21f 100644 --- a/apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts +++ b/apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts @@ -19,6 +19,7 @@ import { RunId, ThreadId, TurnItemId, + knownThreadProjection, type OrchestrationV2DomainEvent, type OrchestrationV2ThreadProjection, type OrchestrationV2TurnItem, @@ -320,7 +321,7 @@ it.layer(TestLayer)("compact bounded snapshot transport", (it) => { "local", ]); } else { - const ids = itemIds(legacy.projection); + const ids = itemIds(knownThreadProjection(legacy.projection)); assert.isTrue(legacy.hasMoreHistory); assert.strictEqual(ids.at(-1), `turn-item:${LONG}:request`); assert.isFalse( diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index b16a2f5b4895a..80e6c753b1d9c 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -4,6 +4,7 @@ import { CommandId, CheckpointId, CheckpointScopeId, + knownThreadProjection, ORCHESTRATION_V2_WS_METHODS, OrchestrationV2CheckpointUnavailableError, WS_METHODS, @@ -263,7 +264,9 @@ const dispatch = (command: OrchestrationV2Command) => request(ORCHESTRATION_V2_WS_METHODS.dispatchCommand, command); const getProjection = (threadId: ThreadId) => - request(ORCHESTRATION_V2_WS_METHODS.getThreadProjection, { threadId }); + request(ORCHESTRATION_V2_WS_METHODS.getThreadProjection, { threadId }).pipe( + Effect.map(knownThreadProjection), + ); const supportsServerResolvedCommandContext = Effect.fn( "EnvironmentCommands.supportsServerResolvedCommandContext", diff --git a/packages/client-runtime/src/platform/orchestrationCache.test.ts b/packages/client-runtime/src/platform/orchestrationCache.test.ts index 64269bf81ca65..2b772e3ad4abf 100644 --- a/packages/client-runtime/src/platform/orchestrationCache.test.ts +++ b/packages/client-runtime/src/platform/orchestrationCache.test.ts @@ -4,6 +4,7 @@ import { EnvironmentId, MessageId, EventId, + knownThreadProjection, TurnItemId, RuntimeRequestId, type OrchestrationV2ShellSnapshot, @@ -108,7 +109,9 @@ describe("orchestration cache envelopes", () => { threadId: v2ThreadId, snapshot: { snapshotSequence: 5, projection }, }); - const decoded = decodeStoredThreadSnapshotJson(encoded).snapshot.projection; + const decoded = knownThreadProjection( + decodeStoredThreadSnapshotJson(encoded).snapshot.projection, + ); expect(decoded.turnItems.find((entry) => entry.id === item.id)).toEqual(item); expect(decoded.visibleTurnItems.find((entry) => entry.item.id === item.id)?.item).toEqual(item); }); diff --git a/packages/client-runtime/src/platform/persistence.ts b/packages/client-runtime/src/platform/persistence.ts index e043279d3e576..703f6bb7777dd 100644 --- a/packages/client-runtime/src/platform/persistence.ts +++ b/packages/client-runtime/src/platform/persistence.ts @@ -2,6 +2,7 @@ import { type EnvironmentId, type OrchestrationV2ShellSnapshot, type OrchestrationV2ThreadDetailSnapshot, + type ReceivedOrchestrationV2ThreadDetailSnapshot, type ServerConfig, type ThreadId, type VcsListRefsResult, @@ -91,7 +92,7 @@ export class EnvironmentCacheStore extends Context.Service< environmentId: EnvironmentId, threadId: ThreadId, ) => Effect.Effect< - Option.Option, + Option.Option, ConnectionPersistenceError >; readonly saveThread: ( diff --git a/packages/client-runtime/src/state/threadHistoryHttp.ts b/packages/client-runtime/src/state/threadHistoryHttp.ts index 2d99d7dbf0e0b..752aac8ed0455 100644 --- a/packages/client-runtime/src/state/threadHistoryHttp.ts +++ b/packages/client-runtime/src/state/threadHistoryHttp.ts @@ -1,4 +1,4 @@ -import type { ThreadId } from "@t3tools/contracts"; +import { knownThreadHistoryPage, type ThreadId } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; @@ -40,5 +40,5 @@ export const fetchEnvironmentThreadHistoryPage = Effect.fn( timeoutMs: input.timeoutMs ?? DEFAULT_THREAD_HISTORY_TIMEOUT_MS, request: ({ client, headers }) => client.threadHistoryPage({ ...endpoint, headers: withOrchestrationProtocolHeader(headers) }), - }); + }).pipe(Effect.map(knownThreadHistoryPage)); }); diff --git a/packages/client-runtime/src/state/threadSnapshotHttp.ts b/packages/client-runtime/src/state/threadSnapshotHttp.ts index f27e2d0a31b6d..be2d4b00d9ee5 100644 --- a/packages/client-runtime/src/state/threadSnapshotHttp.ts +++ b/packages/client-runtime/src/state/threadSnapshotHttp.ts @@ -1,4 +1,8 @@ -import type { OrchestrationV2ThreadDetailSnapshot, ThreadId } from "@t3tools/contracts"; +import { + knownThreadDetailSnapshot, + type OrchestrationV2ThreadDetailSnapshot, + type ThreadId, +} from "@t3tools/contracts"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; @@ -55,7 +59,7 @@ export const fetchEnvironmentThreadSnapshot = Effect.fn( params: { threadId: input.threadId }, headers: withOrchestrationProtocolHeader(headers), }), - }); + }).pipe(Effect.map(knownThreadDetailSnapshot)); }); export type FetchEnvironmentThreadSnapshotError = RemoteEnvironmentRequestError; diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index fc8b8e5008133..44a4e98ae1e5f 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -1,4 +1,5 @@ import { + knownThreadDetailSnapshot, ORCHESTRATION_V2_WS_METHODS, type EnvironmentId as EnvironmentIdType, type OrchestrationV2ThreadDetailSnapshot, @@ -184,6 +185,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const cached = retained === undefined ? yield* cache.loadThread(environmentId, threadId).pipe( + Effect.map(Option.map(knownThreadDetailSnapshot)), Effect.catch((error) => Effect.logWarning("Could not load cached thread.").pipe( Effect.annotateLogs({ diff --git a/packages/contracts/src/baseSchemas.test.ts b/packages/contracts/src/baseSchemas.test.ts index 07ea8b966aecd..25e9864381dcd 100644 --- a/packages/contracts/src/baseSchemas.test.ts +++ b/packages/contracts/src/baseSchemas.test.ts @@ -11,6 +11,7 @@ import { ForwardCompatibleUnion, ForwardCompatibleUnionArray, isUnknownUnionMember, + knownUnionMembers, TrimmedNonEmptyString, UnknownUnionMember, } from "./baseSchemas.ts"; @@ -126,13 +127,35 @@ describe("ForwardCompatibleUnionArray", () => { const decode = fromWire(ForwardCompatibleUnionArray(members, "kind")); - it("drops members a newer server added and keeps the rest", () => { + it("keeps members a newer server added as unknown until a reader drops them", () => { const decoded = decode([ { kind: "triangle" }, { kind: "circle", radius: 1, at }, { kind: "square", side: 2 }, ]); - expect(decoded.map((shape) => shape.kind)).toEqual(["circle", "square"]); + expect(decoded[0]).toEqual(new UnknownUnionMember("kind", "triangle")); + expect(knownUnionMembers(decoded).map((shape) => shape.kind)).toEqual(["circle", "square"]); + }); + + it("returns the same array when every member is known", () => { + const decoded = decode([{ kind: "square", side: 2 }]); + expect(knownUnionMembers(decoded)).toBe(decoded); + }); + + it("validates each known member once", () => { + let checks = 0; + const Counted = Schema.Struct({ + kind: Schema.Literal("counted"), + value: Schema.Number.check( + Schema.makeFilter(() => { + checks++; + return true; + }), + ), + }); + const wire = [{ kind: "counted", value: 1 }, { kind: "future" }, { kind: "counted", value: 2 }]; + expect(fromWire(ForwardCompatibleUnionArray([Counted], "kind"))(wire)).toHaveLength(3); + expect(checks).toBe(2); }); it("fails when a known member is broken, so real bugs stay visible", () => { diff --git a/packages/contracts/src/baseSchemas.ts b/packages/contracts/src/baseSchemas.ts index 60cf9df62c3af..38576f44c8a34 100644 --- a/packages/contracts/src/baseSchemas.ts +++ b/packages/contracts/src/baseSchemas.ts @@ -213,28 +213,25 @@ export const isUnknownUnionMember = ( value: A, ): value is Extract> => value instanceof UnknownUnionMember; +/** + * Keeps members this build does not know as {@link UnknownUnionMember}, so decoding validates each + * known member once. Readers drop them with {@link knownUnionMembers}. + */ export const ForwardCompatibleUnionArray = < const Members extends ReadonlyArray, const Tag extends string, >( members: Members, tag: Tag, -) => - Schema.Array(ForwardCompatibleUnion(members, tag)).pipe( - Schema.decodeTo( - Schema.Array(Schema.toType(Schema.Union(members))), - SchemaTransformation.transform< - ReadonlyArray, - ReadonlyArray> - >({ - decode: (values) => - values.filter((value) => !isUnknownUnionMember(value)) as ReadonlyArray< - Members[number]["Type"] - >, - encode: (values) => values, - }), - ), - ); +) => Schema.Array(ForwardCompatibleUnion(members, tag)); + +/** Returns `values` itself when every member is known. */ +export const knownUnionMembers = ( + values: ReadonlyArray>, +): ReadonlyArray => + (values.some(isUnknownUnionMember) + ? values.filter((value) => !isUnknownUnionMember(value)) + : values) as ReadonlyArray; const knownTags = ( members: ReadonlyArray, diff --git a/packages/contracts/src/orchestrationV2.test.ts b/packages/contracts/src/orchestrationV2.test.ts index 3f4be6092a74c..a0b6777c461b3 100644 --- a/packages/contracts/src/orchestrationV2.test.ts +++ b/packages/contracts/src/orchestrationV2.test.ts @@ -21,6 +21,7 @@ import { ThreadId, TrimmedNonEmptyString, TurnItemId, + UnknownUnionMember, } from "./index.ts"; import { latestProviderTurnForAttempt, @@ -43,6 +44,8 @@ import { OrchestrationV2ThreadShell, OrchestrationV2TurnItem, OrchestrationV2TurnItemJson, + knownThreadHistoryPage, + knownThreadProjection, } from "./orchestrationV2.ts"; const now = DateTime.makeUnsafe("2026-04-20T00:00:00.000Z"); @@ -220,7 +223,7 @@ describe("orchestration V2 contracts", () => { ).toThrow(); }); - it("skips turn item types from a newer server in snapshots and turn-item events", () => { + it("keeps turn item types from a newer server in snapshots until readers drop them", () => { const decodeWireItems = Schema.decodeUnknownSync( Schema.toCodecJson(Schema.Array(OrchestrationV2RpcSchemas.subscribeThread.output)), ); @@ -313,8 +316,11 @@ describe("orchestration V2 contracts", () => { expect(snapshot).toMatchObject({ kind: "snapshot" }); if (snapshot?.kind !== "snapshot") throw new Error("expected a snapshot"); - expect(snapshot.projection.turnItems.map((turnItem) => turnItem.id)).toEqual(["item-known"]); - expect(snapshot.projection.visibleTurnItems.map((row) => row.item.id)).toEqual(["item-known"]); + expect(snapshot.projection.turnItems[0]).toEqual(new UnknownUnionMember("type", "hologram")); + expect(snapshot.projection.visibleTurnItems[0]?.item).toBeInstanceOf(UnknownUnionMember); + const kept = knownThreadProjection(snapshot.projection); + expect(kept.turnItems.map((turnItem) => turnItem.id)).toEqual(["item-known"]); + expect(kept.visibleTurnItems.map((row) => row.item.id)).toEqual(["item-known"]); expect(futureEvent).toEqual({ kind: "unknown-event", sequence: 2, @@ -325,12 +331,14 @@ describe("orchestration V2 contracts", () => { event: { type: "turn-item.updated", payload: { id: "item-known", type: "system_notice" } }, }); expect( - decodeHistoryPage({ - snapshotSequence: 1, - items: [projected(0, future), projected(1, known)], - nextCursor: null, - hasMoreHistory: false, - }).items.map((row) => row.item.id), + knownThreadHistoryPage( + decodeHistoryPage({ + snapshotSequence: 1, + items: [projected(0, future), projected(1, known)], + nextCursor: null, + hasMoreHistory: false, + }), + ).items.map((row) => row.item.id), ).toEqual(["item-known"]); const broken = item("item-broken", "system_notice", {}); @@ -1108,7 +1116,9 @@ describe("orchestration V2 contracts", () => { updatedAt: now, }); - expect(projection.turnItems.map((item) => item.type)).toEqual(["command_execution"]); + expect(knownThreadProjection(projection).turnItems.map((item) => item.type)).toEqual([ + "command_execution", + ]); const boundedSnapshot = decodeOrchestrationV2ThreadStreamItem({ kind: "snapshot", diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 94a9eca331d13..0aa919fea9717 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3,7 +3,6 @@ import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; import * as SchemaAST from "effect/SchemaAST"; import * as SchemaGetter from "effect/SchemaGetter"; -import * as SchemaTransformation from "effect/SchemaTransformation"; import { CheckpointId, @@ -20,6 +19,7 @@ import { ForwardCompatibleUnionArray, hasUnknownUnionTag, isUnknownUnionMember, + knownUnionMembers, NonNegativeInt, PlanId, PositiveInt, @@ -1539,23 +1539,6 @@ const turnItemArray = , ) => ForwardCompatibleUnionArray(union.members, "type"); -const projectedTurnItemArray = ( - row: Row, - rowWithUnknownItem: Item, -) => - Schema.Array(rowWithUnknownItem).pipe( - Schema.decodeTo( - Schema.Array(Schema.toType(row)), - SchemaTransformation.transform, ReadonlyArray>({ - decode: (rows) => - rows.filter( - (projected) => !isUnknownUnionMember((projected as { readonly item: unknown }).item), - ) as ReadonlyArray, - encode: (rows) => rows, - }), - ), - ); - export const OrchestrationV2ProjectedTurnItem = Schema.Struct({ position: NonNegativeInt, visibility: Schema.Literals(["local", "inherited", "synthetic"]), @@ -1565,6 +1548,21 @@ export const OrchestrationV2ProjectedTurnItem = Schema.Struct({ }); export type OrchestrationV2ProjectedTurnItem = typeof OrchestrationV2ProjectedTurnItem.Type; +/** A row whose item a newer server added keeps it as `UnknownUnionMember`. */ +const ReceivedProjectedTurnItem = OrchestrationV2ProjectedTurnItem.mapFields((fields) => ({ + ...fields, + item: ForwardCompatibleUnion(OrchestrationV2TurnItem.members, "type"), +})); +export type ReceivedOrchestrationV2ProjectedTurnItem = typeof ReceivedProjectedTurnItem.Type; + +/** Returns `rows` itself when every row's item is known. */ +export const knownProjectedTurnItems = ( + rows: ReadonlyArray, +): ReadonlyArray => + (rows.some((row) => isUnknownUnionMember(row.item)) + ? rows.filter((row) => !isUnknownUnionMember(row.item)) + : rows) as ReadonlyArray; + export const OrchestrationV2RawProviderEvent = Schema.Struct({ id: RawEventId, driver: ProviderDriverKind, @@ -1753,16 +1751,32 @@ export const OrchestrationV2ThreadProjection = Schema.Struct({ checkpoints: Schema.Array(OrchestrationV2Checkpoint), contextHandoffs: Schema.Array(OrchestrationV2ContextHandoff), contextTransfers: Schema.Array(OrchestrationV2ContextTransfer), - visibleTurnItems: projectedTurnItemArray( - OrchestrationV2ProjectedTurnItem, - OrchestrationV2ProjectedTurnItem.mapFields((fields) => ({ - ...fields, - item: ForwardCompatibleUnion(OrchestrationV2TurnItem.members, "type"), - })), - ), + visibleTurnItems: Schema.Array(ReceivedProjectedTurnItem), updatedAt: Schema.DateTimeUtc, }); -export type OrchestrationV2ThreadProjection = typeof OrchestrationV2ThreadProjection.Type; +/** + * As decoded: turn items a newer server added stay as `UnknownUnionMember`, so each known + * item is validated once. Readers drop them with {@link knownThreadProjection}. + */ +export type ReceivedOrchestrationV2ThreadProjection = typeof OrchestrationV2ThreadProjection.Type; +export interface OrchestrationV2ThreadProjection extends Omit< + ReceivedOrchestrationV2ThreadProjection, + "turnItems" | "visibleTurnItems" +> { + readonly turnItems: ReadonlyArray; + readonly visibleTurnItems: ReadonlyArray; +} + +/** Returns `projection` itself when every turn item is known. */ +export const knownThreadProjection = ( + projection: ReceivedOrchestrationV2ThreadProjection, +): OrchestrationV2ThreadProjection => { + const turnItems = knownUnionMembers(projection.turnItems); + const visibleTurnItems = knownProjectedTurnItems(projection.visibleTurnItems); + return turnItems === projection.turnItems && visibleTurnItems === projection.visibleTurnItems + ? (projection as OrchestrationV2ThreadProjection) + : { ...projection, turnItems, visibleTurnItems }; +}; export const OrchestrationV2ThreadTranscript = Schema.Struct({ threadId: ThreadId, @@ -2316,8 +2330,7 @@ export const OrchestrationV2ThreadProjectionJson = OrchestrationV2ThreadProjecti checkpoints: Schema.Array(OrchestrationV2CheckpointJson), contextHandoffs: Schema.Array(OrchestrationV2ContextHandoffJson), contextTransfers: Schema.Array(OrchestrationV2ContextTransferJson), - visibleTurnItems: projectedTurnItemArray( - OrchestrationV2ProjectedTurnItemJson, + visibleTurnItems: Schema.Array( OrchestrationV2ProjectedTurnItemJson.mapFields((fields) => ({ ...fields, item: ForwardCompatibleUnion(OrchestrationV2TurnItemJson.members, "type"), @@ -3204,7 +3217,24 @@ export const OrchestrationV2ThreadDetailSnapshot = Schema.Struct({ hasMoreHistory: Schema.optionalKey(Schema.Boolean), latestLocalTurnOrdinal: Schema.optionalKey(Schema.NullOr(NonNegativeInt)), }); -export type OrchestrationV2ThreadDetailSnapshot = typeof OrchestrationV2ThreadDetailSnapshot.Type; +/** As decoded; readers drop turn items a newer server added with {@link knownThreadDetailSnapshot}. */ +export type ReceivedOrchestrationV2ThreadDetailSnapshot = + typeof OrchestrationV2ThreadDetailSnapshot.Type; +export interface OrchestrationV2ThreadDetailSnapshot extends Omit< + ReceivedOrchestrationV2ThreadDetailSnapshot, + "projection" +> { + readonly projection: OrchestrationV2ThreadProjection; +} + +export const knownThreadDetailSnapshot = ( + snapshot: ReceivedOrchestrationV2ThreadDetailSnapshot, +): OrchestrationV2ThreadDetailSnapshot => { + const projection = knownThreadProjection(snapshot.projection); + return projection === snapshot.projection + ? (snapshot as OrchestrationV2ThreadDetailSnapshot) + : { ...snapshot, projection }; +}; /** * `historyCursor` is opaque; clients must not parse it. Resume live events with `afterSequence = @@ -3228,17 +3258,25 @@ export type OrchestrationV2ThreadBoundedSnapshot = typeof OrchestrationV2ThreadB export const OrchestrationV2ThreadHistoryPage = Schema.Struct({ snapshotSequence: NonNegativeInt, - items: projectedTurnItemArray( - OrchestrationV2ProjectedTurnItem, - OrchestrationV2ProjectedTurnItem.mapFields((fields) => ({ - ...fields, - item: ForwardCompatibleUnion(OrchestrationV2TurnItem.members, "type"), - })), - ), + items: Schema.Array(ReceivedProjectedTurnItem), nextCursor: Schema.NullOr(TrimmedNonEmptyString), hasMoreHistory: Schema.Boolean, }); -export type OrchestrationV2ThreadHistoryPage = typeof OrchestrationV2ThreadHistoryPage.Type; +/** As decoded; readers drop rows a newer server added with {@link knownThreadHistoryPage}. */ +export type ReceivedOrchestrationV2ThreadHistoryPage = typeof OrchestrationV2ThreadHistoryPage.Type; +export interface OrchestrationV2ThreadHistoryPage extends Omit< + ReceivedOrchestrationV2ThreadHistoryPage, + "items" +> { + readonly items: ReadonlyArray; +} + +export const knownThreadHistoryPage = ( + page: ReceivedOrchestrationV2ThreadHistoryPage, +): OrchestrationV2ThreadHistoryPage => { + const items = knownProjectedTurnItems(page.items); + return items === page.items ? (page as OrchestrationV2ThreadHistoryPage) : { ...page, items }; +}; const knownDomainEventTypes: ReadonlySet = new Set( OrchestrationV2DomainEvent.members.flatMap((member) => { diff --git a/packages/shared/src/orchestrationV2BoundedSnapshot.test.ts b/packages/shared/src/orchestrationV2BoundedSnapshot.test.ts index ad9f7e8b6e38a..24e2feee28fd3 100644 --- a/packages/shared/src/orchestrationV2BoundedSnapshot.test.ts +++ b/packages/shared/src/orchestrationV2BoundedSnapshot.test.ts @@ -2,9 +2,11 @@ import { RunId, ThreadId, TurnItemId, + UnknownUnionMember, type OrchestrationV2ProjectedTurnItem, type OrchestrationV2ThreadProjection, type OrchestrationV2TurnItem, + type ReceivedOrchestrationV2ThreadProjection, } from "@t3tools/contracts"; import { describe, expect, it } from "vite-plus/test"; @@ -105,6 +107,29 @@ describe("compact bounded snapshot turnItems", () => { ).toEqual(JSON.parse(JSON.stringify(full))); }); + it("drops turn items a newer server added before restoring local rows", () => { + const { projection: full } = forkWindow(); + const compact = omitLocalVisibleTurnItems(full)!; + const future = new UnknownUnionMember("type", "hologram"); + const received = { + ...compact, + turnItems: [future, ...compact.turnItems], + visibleTurnItems: [ + ...compact.visibleTurnItems, + { + position: 7, + visibility: "local", + sourceThreadId: CHILD, + sourceItemId: "future", + item: future, + }, + ], + } as unknown as ReceivedOrchestrationV2ThreadProjection; + expect( + boundedSnapshotProjection({ projection: received, turnItemsOmitLocalVisible: true }), + ).toEqual(full); + }); + it("treats rows sourced from the thread itself as local", () => { const own = item(CHILD, "own", 1); const full = projection({ visible: [row("inherited", CHILD, own)], turnItems: [own] }); diff --git a/packages/shared/src/orchestrationV2BoundedSnapshot.ts b/packages/shared/src/orchestrationV2BoundedSnapshot.ts index 599ccd5117e9c..0e330377ae914 100644 --- a/packages/shared/src/orchestrationV2BoundedSnapshot.ts +++ b/packages/shared/src/orchestrationV2BoundedSnapshot.ts @@ -1,6 +1,8 @@ -import type { - OrchestrationV2ProjectedTurnItem, - OrchestrationV2ThreadProjection, +import { + knownThreadProjection, + type OrchestrationV2ProjectedTurnItem, + type OrchestrationV2ThreadProjection, + type ReceivedOrchestrationV2ThreadProjection, } from "@t3tools/contracts"; function isLocalTimelineRow( @@ -39,11 +41,13 @@ export function restoreLocalVisibleTurnItems( return { ...projection, turnItems: [...local, ...projection.turnItems] }; } +/** Also drops the turn items a newer server added, which this build does not know. */ export function boundedSnapshotProjection(snapshot: { - readonly projection: OrchestrationV2ThreadProjection; + readonly projection: ReceivedOrchestrationV2ThreadProjection; readonly turnItemsOmitLocalVisible?: true | undefined; }): OrchestrationV2ThreadProjection { + const projection = knownThreadProjection(snapshot.projection); return snapshot.turnItemsOmitLocalVisible === true - ? restoreLocalVisibleTurnItems(snapshot.projection) - : snapshot.projection; + ? restoreLocalVisibleTurnItems(projection) + : projection; } From 245aa4f512823a7cc56b64cc049bb237dd35456a Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Mon, 19 Oct 2026 10:18:38 -0500 Subject: [PATCH 2/6] fix(contracts): keep knownProjectedTurnItems module-internal Knip flagged the export as unused; only knownThreadProjection and knownThreadHistoryPage call it. Co-Authored-By: Claude Opus 5.5 --- packages/contracts/src/orchestrationV2.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 0aa919fea9717..ff3e174dac175 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1556,7 +1556,7 @@ const ReceivedProjectedTurnItem = OrchestrationV2ProjectedTurnItem.mapFields((fi export type ReceivedOrchestrationV2ProjectedTurnItem = typeof ReceivedProjectedTurnItem.Type; /** Returns `rows` itself when every row's item is known. */ -export const knownProjectedTurnItems = ( +const knownProjectedTurnItems = ( rows: ReadonlyArray, ): ReadonlyArray => (rows.some((row) => isUnknownUnionMember(row.item)) From c00f367b08287062c97d5e5f7d6f2a7b08c3dac9 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 19:24:56 +0000 Subject: [PATCH 3/6] test(client): cover timeline filtering at HTTP readers --- .../src/state/environmentHttpAuth.test.ts | 32 +++++++++++++++++++ packages/contracts/src/orchestrationV2.ts | 3 +- 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/packages/client-runtime/src/state/environmentHttpAuth.test.ts b/packages/client-runtime/src/state/environmentHttpAuth.test.ts index 4cd0ee805667f..088473d4e9ba2 100644 --- a/packages/client-runtime/src/state/environmentHttpAuth.test.ts +++ b/packages/client-runtime/src/state/environmentHttpAuth.test.ts @@ -259,6 +259,38 @@ const LOADERS: ReadonlyArray<{ ]; describe("authenticated environment HTTP requests", () => { + it.effect("filters future timeline items at HTTP read boundaries", () => + Effect.gen(function* () { + const future = { type: "future_item" }; + const row = { + position: 0, + visibility: "local", + sourceThreadId: THREAD.projection.thread.id, + sourceItemId: "future-item", + item: future, + }; + for (const loader of LOADERS.filter((entry) => + ["thread snapshot", "bounded thread snapshot", "older thread history"].includes(entry.name), + )) { + const wire = loader.response as Record; + const response = + "projection" in wire + ? { + ...wire, + projection: { + ...(wire.projection as Record), + turnItems: [future], + visibleTurnItems: [row], + }, + } + : { ...wire, items: [row] }; + const harness = makeHarness(() => Response.json(response)); + const result = yield* loader.load(harness.input).pipe(Effect.provide(harness.httpLayer)); + expect(result).toEqual(loader.expected); + } + }), + ); + it.effect.each(LOADERS)("rejects an invalid $name response", (loader) => Effect.gen(function* () { const harness = makeHarness(() => Response.json({})); diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index e0ef00a0d9042..e833a63c030c8 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1679,11 +1679,10 @@ const ReceivedProjectedTurnItem = OrchestrationV2ProjectedTurnItem.mapFields((fi ...fields, item: ForwardCompatibleUnion(OrchestrationV2TurnItem.members, "type"), })); -export type ReceivedOrchestrationV2ProjectedTurnItem = typeof ReceivedProjectedTurnItem.Type; /** Returns `rows` itself when every row's item is known. */ const knownProjectedTurnItems = ( - rows: ReadonlyArray, + rows: ReadonlyArray, ): ReadonlyArray => (rows.some((row) => isUnknownUnionMember(row.item)) ? rows.filter((row) => !isUnknownUnionMember(row.item)) From ffe511923cd328ee49b7a7d26882a8b49bb3168c Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 19:33:12 +0000 Subject: [PATCH 4/6] fix(client): filter compact cache reads before restoration --- .../environment-cache-store.test.ts | 3 ++- .../src/guest/GuestShareService.test.ts | 3 ++- .../src/platform/orchestrationCache.test.ts | 24 +++++++++++++++++++ .../src/platform/threadCacheEncoding.ts | 13 ++++++---- 4 files changed, 36 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 d8031de9e7a39..677291981203f 100644 --- a/apps/mobile/src/connection/environment-cache-store.test.ts +++ b/apps/mobile/src/connection/environment-cache-store.test.ts @@ -6,6 +6,7 @@ import { } from "@t3tools/client-runtime/platform"; import { CommandId, + knownThreadDetailSnapshot, EnvironmentId, ProjectId, ProviderInstanceId, @@ -277,7 +278,7 @@ describe("mobile SQLite environment cache store", () => { const loaded = Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)); expect(loaded).toEqual(snapshot); expect(memory.values.get(id)).toBe(raw); - yield* store.saveThread(ENVIRONMENT_ID, loaded); + yield* store.saveThread(ENVIRONMENT_ID, knownThreadDetailSnapshot(loaded)); expect(memory.schemaVersions.get(id)).toBe(COMPACT_THREAD_CACHE_SCHEMA_VERSION); expect(Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID))).toEqual( snapshot, diff --git a/apps/server/src/guest/GuestShareService.test.ts b/apps/server/src/guest/GuestShareService.test.ts index 2f0366d4d2d22..14894ef380062 100644 --- a/apps/server/src/guest/GuestShareService.test.ts +++ b/apps/server/src/guest/GuestShareService.test.ts @@ -5,6 +5,7 @@ import { CommandId, EventId, GuestShareId, + knownThreadProjection, MessageId, type OrchestrationV2TurnItem, ProjectId, @@ -267,7 +268,7 @@ it.effect("starts a stream with the runs and items already in the thread", () => } assert.equal(snapshot.item.projection.runs.length, 1); assert.include( - snapshot.item.projection.visibleTurnItems.map((row) => row.item.type), + knownThreadProjection(snapshot.item.projection).visibleTurnItems.map((row) => row.item.type), "user_message", ); }).pipe(Effect.provide(layerTest)), diff --git a/packages/client-runtime/src/platform/orchestrationCache.test.ts b/packages/client-runtime/src/platform/orchestrationCache.test.ts index 1196046459884..af20d3a011331 100644 --- a/packages/client-runtime/src/platform/orchestrationCache.test.ts +++ b/packages/client-runtime/src/platform/orchestrationCache.test.ts @@ -18,6 +18,8 @@ import { v2ThreadId, } from "../state/orchestrationV2TestFixtures.ts"; +import { decodeStoredThreadSnapshot, encodeStoredThreadSnapshot } from "./threadCacheEncoding.ts"; + import { applyOrchestrationV2ProjectionEvent } from "../state/orchestrationV2Projection.ts"; const environmentId = EnvironmentId.make("environment-cache-test"); @@ -35,6 +37,28 @@ class TestCacheDecodeError extends Schema.TaggedError()( ) {} describe("orchestration cache envelopes", () => { + it.effect("discards compact caches with unknown items without weakening count checks", () => + Effect.gen(function* () { + const stored = { + environmentId, + threadId: v2ThreadId, + snapshot: { snapshotSequence: 4, projection: v2Projection }, + }; + const encoded = yield* encodeStoredThreadSnapshot(stored); + expect((yield* decodeStoredThreadSnapshot(encoded)).snapshot.projection).toEqual( + v2Projection, + ); + const wire = JSON.parse(encoded); + wire.snapshot.projection.turnItems = [{ type: "future_item" }]; + wire.snapshot.turnItemCount = 1; + const unknown = yield* Effect.flip(decodeStoredThreadSnapshot(JSON.stringify(wire))); + expect(unknown._tag).toBe("InvalidThreadCache"); + wire.snapshot.projection.turnItems = []; + const truncated = yield* Effect.flip(decodeStoredThreadSnapshot(JSON.stringify(wire))); + expect(truncated._tag).toBe("InvalidThreadCache"); + }), + ); + it("preserves live provider notices through persisted thread snapshots", () => { const item = { id: TurnItemId.make("notice-cache"), diff --git a/packages/client-runtime/src/platform/threadCacheEncoding.ts b/packages/client-runtime/src/platform/threadCacheEncoding.ts index b7c233942de43..e8573ab187b51 100644 --- a/packages/client-runtime/src/platform/threadCacheEncoding.ts +++ b/packages/client-runtime/src/platform/threadCacheEncoding.ts @@ -1,3 +1,7 @@ +import { + knownThreadProjection, + type OrchestrationV2ThreadDetailSnapshot, +} from "@t3tools/contracts"; import { omitLocalVisibleTurnItems, restoreLocalVisibleTurnItems, @@ -33,10 +37,9 @@ const decodeCompatible = Schema.decodeUnknownEffect( ); export const encodeStoredThreadSnapshot = Effect.fn("ThreadCache.encode")(function* ( - stored: Pick< - typeof StoredOrchestrationThreadSnapshot.Type, - "environmentId" | "threadId" | "snapshot" - >, + stored: Pick & { + readonly snapshot: OrchestrationV2ThreadDetailSnapshot; + }, ) { const { snapshot } = stored; const compact = omitLocalVisibleTurnItems(snapshot.projection); @@ -58,7 +61,7 @@ export const decodeStoredThreadSnapshot = Effect.fn("ThreadCache.decode")(functi if (stored.schemaVersion !== COMPACT_THREAD_CACHE_SCHEMA_VERSION) return stored; const { turnItemsOmitLocalVisible, turnItemCount, visibleTurnItemCount, ...snapshot } = stored.snapshot; - const { projection } = snapshot; + const projection = knownThreadProjection(snapshot.projection); const threadId = projection.thread.id; let previousPosition = -1; const visibleIds = new Set(); From 6642dbdb1012a59970c18fa258063f1a484c469e Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 30 Oct 2026 00:05:59 +0000 Subject: [PATCH 5/6] fix(scripts): filter received timeline projections --- scripts/mobile-behavior/environment.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/scripts/mobile-behavior/environment.ts b/scripts/mobile-behavior/environment.ts index aa526a1d943e0..bbc6c6a73e2a0 100644 --- a/scripts/mobile-behavior/environment.ts +++ b/scripts/mobile-behavior/environment.ts @@ -13,6 +13,7 @@ import { EnvironmentId, ExecutionEnvironmentDescriptor, MessageId, + knownThreadProjection, ORCHESTRATION_PROTOCOL_HEADER, ORCHESTRATION_PROTOCOL_VERSION, OrchestrationV2ShellSnapshot, @@ -234,7 +235,7 @@ export const startEnvironment = Effect.fn("environment.start")(function* (input: const command = Effect.map(uuid, CommandId.make); const projection = (threadId: ThreadId) => - client["orchestration.getThreadProjection"]({ threadId }); + client["orchestration.getThreadProjection"]({ threadId }).pipe(Effect.map(knownThreadProjection)); const create = Effect.fn("environment.create")(function* ( title: string, From 7e40a453d76d5de2a253feb60112b3b3055df765 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 30 Oct 2026 00:13:32 +0000 Subject: [PATCH 6/6] style(scripts): format timeline projection reader --- scripts/mobile-behavior/environment.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/scripts/mobile-behavior/environment.ts b/scripts/mobile-behavior/environment.ts index bbc6c6a73e2a0..54979c1b0b216 100644 --- a/scripts/mobile-behavior/environment.ts +++ b/scripts/mobile-behavior/environment.ts @@ -235,7 +235,9 @@ export const startEnvironment = Effect.fn("environment.start")(function* (input: const command = Effect.map(uuid, CommandId.make); const projection = (threadId: ThreadId) => - client["orchestration.getThreadProjection"]({ threadId }).pipe(Effect.map(knownThreadProjection)); + client["orchestration.getThreadProjection"]({ threadId }).pipe( + Effect.map(knownThreadProjection), + ); const create = Effect.fn("environment.create")(function* ( title: string,