perf(contracts): filter unknown timeline members at read boundaries

apps/mobile/scripts/measure-renders.mts

@@ -14,7 +14,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";
@@ -223,10 +225,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),
@@ -243,7 +249,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);
 

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,

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;

apps/server/src/guest/GuestShareService.test.ts

@@ -5,6 +5,7 @@ import {
   CommandId,
   EventId,
   GuestShareId,
+  knownThreadProjection,
   type GuestStreamEvent,
   MessageId,
   type OrchestrationV2TurnItem,
@@ -271,7 +272,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)),

apps/server/src/orchestration-v2/ThreadStream.ts

@@ -35,10 +35,11 @@ export function threadReplayEncodedBytes(items: ReadonlyArray<unknown>): 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<OrchestrationV2ThreadStreamItem, { readonly kind: "snapshot" }>,
+  "projection"
+> & { readonly projection: OrchestrationV2ThreadProjection };
 
 export function boundedSnapshotResponseFields(input: {
   readonly bounded: BoundedProjectionResult;

apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts

@@ -23,6 +23,7 @@ import {
   RunId,
   ThreadId,
   TurnItemId,
+  knownThreadProjection,
   type OrchestrationV2DomainEvent,
   type OrchestrationV2StoredEvent,
   type OrchestrationV2Subagent,
@@ -488,7 +489,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(

packages/client-runtime/src/operations/commands.ts

@@ -4,6 +4,7 @@ import {
   CommandId,
   CheckpointId,
   CheckpointScopeId,
+  knownThreadProjection,
   ORCHESTRATION_V2_WS_METHODS,
   OrchestrationV2CheckpointUnavailableError,
   WS_METHODS,
@@ -267,7 +268,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 persistAttachments = Effect.fn("EnvironmentCommands.persistAttachments")(function* (
   threadId: ThreadId,

packages/client-runtime/src/platform/orchestrationCache.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from "@effect/vitest";
-import { EnvironmentId, EventId, TurnItemId } from "@t3tools/contracts";
+import { EnvironmentId, EventId, TurnItemId, knownThreadProjection } from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
 import * as Option from "effect/Option";
 import * as Schema from "effect/Schema";
@@ -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<TestCacheDecodeError>()(
 ) {}
 
 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"),
@@ -69,7 +93,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);
   });

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<OrchestrationV2ThreadDetailSnapshot>,
+      Option.Option<ReceivedOrchestrationV2ThreadDetailSnapshot>,
       ConnectionPersistenceError
     >;
     readonly saveThread: (

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<typeof StoredOrchestrationThreadSnapshot.Type, "environmentId" | "threadId"> & {
+    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<string>();

packages/client-runtime/src/state/environmentHttpAuth.test.ts

@@ -258,6 +258,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) =>
+        ["bounded thread snapshot", "older thread history"].includes(entry.name),
+      )) {
+        const wire = loader.response as Record<string, unknown>;
+        const response =
+          "projection" in wire
+            ? {
+                ...wire,
+                projection: {
+                  ...(wire.projection as Record<string, unknown>),
+                  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({}));

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";
 
@@ -42,5 +42,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));
 });

packages/client-runtime/src/state/threads.ts

@@ -1,5 +1,6 @@
 import {
   isOrchestrationV2WorkActive,
+  knownThreadDetailSnapshot,
   ORCHESTRATION_V2_WS_METHODS,
   type EnvironmentId as EnvironmentIdType,
   type OrchestrationV2ThreadDetailSnapshot,
@@ -211,6 +212,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({
@@ -869,9 +871,10 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
           const saved =
             queuedSnapshot ??
             Option.getOrUndefined(
-              yield* cache
-                .loadThread(environmentId, threadId)
-                .pipe(Effect.orElseSucceed(() => Option.none())),
+              yield* cache.loadThread(environmentId, threadId).pipe(
+                Effect.map(Option.map(knownThreadDetailSnapshot)),
+                Effect.orElseSucceed(() => Option.none()),
+              ),
             );
           if (saved === undefined) return;
           const projection = settleTurnItems(threadId, saved.projection, latestItems);

packages/contracts/src/baseSchemas.test.ts

@@ -7,6 +7,7 @@ import {
   ForwardCompatibleUnion,
   ForwardCompatibleUnionArray,
   isUnknownUnionMember,
+  knownUnionMembers,
   TrimmedNonEmptyString,
   UnknownUnionMember,
 } from "./baseSchemas.ts";
@@ -85,13 +86,36 @@ 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 }];
+    const decoded = fromWire(ForwardCompatibleUnionArray([Counted], "kind"))(wire);
+    expect(checks).toBe(2);
+    expect(decoded).toHaveLength(3);
   });
 
   it("fails when a known member is broken, so real bugs stay visible", () => {