perf(web): a thread loaded from the cache holds each item once

apps/web/src/connection/storage.ts

@@ -4,8 +4,8 @@ import {
   EMPTY_CONNECTION_CATALOG_DOCUMENT,
   ORCHESTRATION_CACHE_SCHEMA_VERSION,
   StoredOrchestrationShellSnapshot,
-  StoredOrchestrationThreadSnapshot,
   decodeOrDiscardOrchestrationCache,
+  decodeStoredThreadSnapshot,
   putRemoteDpopTokenInCatalog,
   registerConnectionInCatalog,
   removeCatalogValue,
@@ -64,8 +64,6 @@ const QUERY_PRUNE_BATCH_BYTES = 1024 * 1024;
 const CATALOG_KEY = "document";
 const StoredShellSnapshot = StoredOrchestrationShellSnapshot;
 const StoredShellSnapshotJson = Schema.fromJsonString(StoredShellSnapshot);
-const StoredThreadSnapshot = StoredOrchestrationThreadSnapshot;
-const StoredThreadSnapshotJson = Schema.fromJsonString(StoredThreadSnapshot);
 const StoredServerConfig = Schema.Struct({
   schemaVersion: Schema.Literal(1),
   environmentId: EnvironmentId,
@@ -87,7 +85,6 @@ const ConnectionCatalogDocumentJson = Schema.fromJsonString(ConnectionCatalogDoc
 const decodeConnectionCatalogDocument = Schema.decodeUnknownEffect(ConnectionCatalogDocumentJson);
 const encodeConnectionCatalogDocument = Schema.encodeEffect(ConnectionCatalogDocumentJson);
 const decodeStoredShellSnapshot = Schema.decodeUnknownEffect(StoredShellSnapshotJson);
-const decodeStoredThreadSnapshot = Schema.decodeUnknownEffect(StoredThreadSnapshotJson);
 const decodeStoredServerConfig = Schema.decodeUnknownEffect(StoredServerConfigJson);
 const encodeStoredServerConfig = Schema.encodeEffect(StoredServerConfigJson);
 const decodeStoredVcsRefs = Schema.decodeUnknownEffect(StoredVcsRefsJson);
@@ -724,7 +721,7 @@ export const layer = Layer.effectContext(
       yieldToHost: yieldToBrowserTask,
     });
     const encodeStoredThreadSnapshot = makeStoredThreadSnapshotEncoder({
-      format: "full",
+      format: "compact",
       yieldToHost: yieldToBrowserTask,
     });
     const threadWriteOrder = makeThreadCacheWriteOrder();

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

@@ -1,8 +1,10 @@
 import { describe, expect, it } from "@effect/vitest";
 import {
   EnvironmentId,
+  MessageId,
   ThreadId,
   TurnItemId,
+  type OrchestrationV2ThreadDetailSnapshot,
   type OrchestrationV2ThreadProjection,
 } from "@t3tools/contracts";
 import * as DateTime from "effect/DateTime";
@@ -139,3 +141,136 @@ describe("cooperative thread cache encoding", () => {
     }),
   );
 });
+
+describe("thread cache records", () => {
+  const BODY = "a body long enough to count ".repeat(4);
+  const COMMAND_OUTPUT = "only the items carry this output";
+
+  const base = (id: string, ordinal: number) => ({
+    id: TurnItemId.make(id),
+    threadId: v2ThreadId,
+    runId: null,
+    nodeId: null,
+    providerThreadId: null,
+    providerTurnId: null,
+    nativeItemRef: null,
+    parentItemId: null,
+    ordinal,
+    status: "completed" as const,
+    title: null,
+    startedAt: v2Now,
+    completedAt: v2Now,
+    updatedAt: v2Now,
+  });
+
+  const messageId = MessageId.make("message-1");
+  const answer = {
+    ...base("answer", 1),
+    type: "assistant_message" as const,
+    messageId,
+    text: BODY,
+    streaming: false,
+  };
+  const command = {
+    ...base("command", 2),
+    type: "command_execution" as const,
+    input: "pwd",
+    output: COMMAND_OUTPUT,
+    exitCode: 0,
+  };
+  const rowOf = <Item extends { readonly id: TurnItemId }>(item: Item, position: number) => ({
+    position,
+    visibility: "local" as const,
+    sourceThreadId: v2ThreadId,
+    sourceItemId: item.id,
+    item,
+  });
+  const bodyProjection: OrchestrationV2ThreadProjection = {
+    ...v2Projection,
+    messages: [
+      {
+        id: messageId,
+        threadId: v2ThreadId,
+        runId: null,
+        nodeId: null,
+        role: "assistant",
+        text: BODY,
+        attachments: [],
+        streaming: false,
+        createdBy: "agent",
+        creationSource: "provider",
+        createdAt: v2Now,
+        updatedAt: v2Now,
+      },
+    ],
+    turnItems: [answer, command],
+    visibleTurnItems: [rowOf(answer, 0), rowOf(command, 1)],
+  };
+  const snapshot: OrchestrationV2ThreadDetailSnapshot = {
+    snapshotSequence: 3,
+    projection: bodyProjection,
+  };
+  const record = { environmentId, threadId: v2ThreadId, snapshot };
+
+  const encodeStoredThreadSnapshot = (input: typeof record) =>
+    makeStoredThreadSnapshotEncoder({ format: "compact" })(input).pipe(
+      Effect.map(Option.getOrThrow),
+    );
+
+  it.effect("shares the copies of a record that could not be compacted", () =>
+    Effect.gen(function* () {
+      // A hidden item between two visible ones: `turnItems` no longer starts with the visible rows.
+      const hidden = {
+        ...base("hidden", 3),
+        type: "reasoning" as const,
+        text: "not shown",
+        streaming: false,
+      };
+      const uncompacted = {
+        ...bodyProjection,
+        turnItems: [answer, hidden, command],
+      };
+      const raw = yield* encodeStoredThreadSnapshot({
+        ...record,
+        snapshot: { ...snapshot, projection: uncompacted },
+      });
+      expect(JSON.parse(raw).snapshot.turnItemsOmitLocalVisible).toBe(false);
+      const { projection: loaded } = (yield* decodeStoredThreadSnapshot(raw)).snapshot;
+      expect(loaded).toEqual(uncompacted);
+      expect(loaded.visibleTurnItems[0]!.item).toBe(loaded.turnItems[0]);
+      expect(loaded.visibleTurnItems[1]!.item).toBe(loaded.turnItems[2]);
+    }),
+  );
+
+  it.effect("shares the matching copies of a full record", () =>
+    Effect.gen(function* () {
+      const raw = yield* encodeOriginal({
+        schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
+        ...record,
+      });
+      const { projection: loaded } = (yield* decodeStoredThreadSnapshot(raw)).snapshot;
+      expect(loaded).toEqual(bodyProjection);
+      expect(loaded.visibleTurnItems[1]!.item).toBe(loaded.turnItems[1]);
+    }),
+  );
+
+  it.effect("keeps both copies of a full record when they differ", () =>
+    Effect.gen(function* () {
+      const stale = { ...command, output: "an older output" };
+      const raw = yield* encodeOriginal({
+        schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
+        ...record,
+        snapshot: {
+          ...snapshot,
+          projection: {
+            ...bodyProjection,
+            visibleTurnItems: [rowOf(answer, 0), rowOf(stale, 1)],
+          },
+        },
+      });
+      const { projection: loaded } = (yield* decodeStoredThreadSnapshot(raw)).snapshot;
+      expect(loaded.turnItems[1]).toMatchObject({ output: COMMAND_OUTPUT });
+      expect(loaded.visibleTurnItems[1]!.item).toMatchObject({ output: "an older output" });
+    }),
+  );
+});

packages/client-runtime/src/platform/threadCacheEncoding.ts

@@ -1,9 +1,17 @@
-import { OrchestrationV2ThreadProjectionJson } from "@t3tools/contracts";
+import {
+  type OrchestrationV2ThreadProjection,
+  OrchestrationV2ThreadProjectionJson,
+} from "@t3tools/contracts";
 import {
   omitLocalVisibleTurnItems,
   restoreLocalVisibleTurnItems,
 } from "@t3tools/shared/orchestrationV2BoundedSnapshot";
+import {
+  dedupeSnapshotBodies,
+  restoreSnapshotBodies,
+} from "@t3tools/shared/orchestrationV2DedupedBodies";
 import * as Effect from "effect/Effect";
+import * as Equal from "effect/Equal";
 import * as Option from "effect/Option";
 import * as Schema from "effect/Schema";
 
@@ -13,7 +21,8 @@ import {
   StoredOrchestrationThreadSnapshot,
 } from "./orchestrationCache.ts";
 
-// Keep the full format for web/desktop; older clients discard compact records.
+// Each item is written once and shared again on decode, so a thread loaded from the cache costs
+// what a thread from the server costs. Older clients discard these records and refetch.
 export const COMPACT_THREAD_CACHE_SCHEMA_VERSION = 6 as const;
 
 /**
@@ -41,6 +50,25 @@ class InvalidThreadCache extends Schema.TaggedError<InvalidThreadCache>()(
   {},
 ) {}
 
+/** JSON decodes every copy of a body separately: point an item at its record's copy again. */
+function shareBodiesWithRecords(
+  projection: OrchestrationV2ThreadProjection,
+): OrchestrationV2ThreadProjection {
+  return restoreSnapshotBodies(dedupeSnapshotBodies(projection));
+}
+
+/** A full record decodes both copies of each item: keep one object wherever they match. */
+function shareFullRecordItems(
+  projection: OrchestrationV2ThreadProjection,
+): OrchestrationV2ThreadProjection {
+  const items = new Map(projection.turnItems.map((item) => [item.id, item]));
+  const visibleTurnItems = projection.visibleTurnItems.map((row) => {
+    const item = row.sourceThreadId === projection.thread.id ? items.get(row.item.id) : undefined;
+    return item !== undefined && Equal.equals(item, row.item) ? { ...row, item } : row;
+  });
+  return { ...projection, visibleTurnItems };
+}
+
 const decodeCompatible = Schema.decodeUnknownEffect(
   Schema.fromJsonString(
     Schema.Union([StoredCompactThreadSnapshot, StoredOrchestrationThreadSnapshot]),
@@ -222,7 +250,15 @@ export function makeStoredThreadSnapshotEncoder(options: {
 
 export const decodeStoredThreadSnapshot = Effect.fn("ThreadCache.decode")(function* (raw: string) {
   const stored = yield* decodeCompatible(raw);
-  if (stored.schemaVersion !== COMPACT_THREAD_CACHE_SCHEMA_VERSION) return stored;
+  if (stored.schemaVersion === ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION) {
+    return {
+      ...stored,
+      snapshot: {
+        ...stored.snapshot,
+        projection: shareBodiesWithRecords(shareFullRecordItems(stored.snapshot.projection)),
+      },
+    };
+  }
   const { turnItemsOmitLocalVisible, turnItemCount, visibleTurnItemCount, ...snapshot } =
     stored.snapshot;
   const { projection } = snapshot;
@@ -245,9 +281,11 @@ export const decodeStoredThreadSnapshot = Effect.fn("ThreadCache.decode")(functi
     visibleIds.add(identity);
     return valid;
   });
+  // A record that could not be compacted carries both copies: a retained request card followed by
+  // newer items is enough, since `turnItems` then no longer starts with the visible rows.
   const restored = turnItemsOmitLocalVisible
     ? restoreLocalVisibleTurnItems(projection)
-    : projection;
+    : shareFullRecordItems(projection);
   const itemIds = new Set<string>();
   const validItems = restored.turnItems.every((item) => {
     const valid = item.threadId === threadId && !itemIds.has(item.id);
@@ -266,5 +304,5 @@ export const decodeStoredThreadSnapshot = Effect.fn("ThreadCache.decode")(functi
   ) {
     return yield* new InvalidThreadCache();
   }
-  return { ...stored, snapshot: { ...snapshot, projection: restored } };
+  return { ...stored, snapshot: { ...snapshot, projection: shareBodiesWithRecords(restored) } };
 });