perf(mobile): compact persisted thread snapshots

apps/mobile/src/connection/environment-cache-store.test.ts

@@ -1,13 +1,17 @@
 import {
   ORCHESTRATION_CACHE_SCHEMA_VERSION,
   ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
+  COMPACT_THREAD_CACHE_SCHEMA_VERSION,
+  StoredOrchestrationThreadSnapshot,
 } from "@t3tools/client-runtime/platform";
 import {
   CommandId,
   EnvironmentId,
   ProjectId,
   ProviderInstanceId,
+  RunId,
   ThreadId,
+  TurnItemId,
   type OrchestrationV2ShellSnapshot,
   type OrchestrationV2ThreadDetailSnapshot,
   type ThreadPullRequestLink,
@@ -17,6 +21,7 @@ import * as DateTime from "effect/DateTime";
 import * as Effect from "effect/Effect";
 import * as Logger from "effect/Logger";
 import * as Option from "effect/Option";
+import * as Schema from "effect/Schema";
 
 import * as MobileDatabase from "../persistence/mobile-database";
 import { make } from "./environment-cache-store";
@@ -26,6 +31,9 @@ const PROJECT_ID = ProjectId.make("project-1");
 const THREAD_ID = ThreadId.make("thread-1");
 const PROVIDER_INSTANCE_ID = ProviderInstanceId.make("codex");
 const NOW = DateTime.makeUnsafe("2026-07-29T12:00:00.000Z");
+const encodeLegacyThreadSnapshot = Schema.encodeEffect(
+  Schema.fromJsonString(StoredOrchestrationThreadSnapshot),
+);
 const SHELL_SNAPSHOT: OrchestrationV2ShellSnapshot = {
   schemaVersion: 1,
   snapshotSequence: 3,
@@ -152,6 +160,59 @@ function cacheId(
   return `${environmentId}:${kind}:${cacheKey}`;
 }
 
+function threadSnapshotWithItems(): OrchestrationV2ThreadDetailSnapshot {
+  const local = {
+    id: TurnItemId.make("local-notice"),
+    threadId: THREAD_ID,
+    runId: null,
+    nodeId: null,
+    providerThreadId: null,
+    providerTurnId: null,
+    nativeItemRef: null,
+    parentItemId: null,
+    ordinal: 10,
+    status: "completed" as const,
+    title: null,
+    startedAt: NOW,
+    completedAt: NOW,
+    updatedAt: NOW,
+    type: "system_notice" as const,
+    message: "A settled local notice.",
+  };
+  const hidden = { ...local, id: TurnItemId.make("hidden-notice"), ordinal: 11 };
+  const inherited = {
+    ...local,
+    id: TurnItemId.make("inherited-notice"),
+    threadId: ThreadId.make("parent-thread"),
+  };
+  return {
+    ...THREAD_SNAPSHOT,
+    historyCursor: "older-page",
+    hasMoreHistory: true,
+    latestLocalTurnOrdinal: 11,
+    projection: {
+      ...THREAD_SNAPSHOT.projection,
+      turnItems: [local, hidden],
+      visibleTurnItems: [
+        {
+          position: 0,
+          visibility: "inherited",
+          sourceThreadId: inherited.threadId,
+          sourceItemId: inherited.id,
+          item: inherited,
+        },
+        {
+          position: 1,
+          visibility: "local",
+          sourceThreadId: THREAD_ID,
+          sourceItemId: local.id,
+          item: local,
+        },
+      ],
+    },
+  };
+}
+
 function makeDatabase() {
   const values = new Map<string, string>();
   const schemaVersions = new Map<string, number>();
@@ -198,6 +259,238 @@ function makeDatabase() {
 }
 
 describe("mobile SQLite environment cache store", () => {
+  it.effect("loads a V5 record and migrates to V6 on the next save without losing items", () =>
+    Effect.gen(function* () {
+      const memory = makeDatabase();
+      const store = yield* make().pipe(
+        Effect.provideService(MobileDatabase.MobileDatabase, memory.database),
+      );
+      const snapshot = threadSnapshotWithItems();
+      const id = cacheId(ENVIRONMENT_ID, "thread", THREAD_ID);
+      const raw = yield* encodeLegacyThreadSnapshot({
+        schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
+        environmentId: ENVIRONMENT_ID,
+        threadId: THREAD_ID,
+        snapshot,
+      });
+      memory.values.set(id, raw);
+      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);
+      expect(memory.schemaVersions.get(id)).toBe(COMPACT_THREAD_CACHE_SCHEMA_VERSION);
+      expect(Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID))).toEqual(
+        snapshot,
+      );
+      expect(memory.removed).toEqual([]);
+    }),
+  );
+
+  it.effect("discards corrupt or partial thread records so they can be refetched", () =>
+    Effect.gen(function* () {
+      const memory = makeDatabase();
+      const store = yield* make().pipe(
+        Effect.provideService(MobileDatabase.MobileDatabase, memory.database),
+      );
+      const id = cacheId(ENVIRONMENT_ID, "thread", THREAD_ID);
+      yield* store.saveThread(ENVIRONMENT_ID, threadSnapshotWithItems());
+      const raw = memory.values.get(id)!;
+      const partial = JSON.parse(raw);
+      delete partial.snapshot.projection.visibleTurnItems;
+      const invalidItem = JSON.parse(raw);
+      delete invalidItem.snapshot.projection.visibleTurnItems[1].item.message;
+      const corruptions = [
+        partial,
+        invalidItem,
+        ...[
+          (value: typeof partial) => {
+            delete value.snapshot.turnItemCount;
+          },
+          (value: typeof partial) => {
+            delete value.snapshot.turnItemsOmitLocalVisible;
+          },
+          (value: typeof partial) => {
+            value.snapshot.turnItemsOmitLocalVisible = false;
+          },
+          (value: typeof partial) => {
+            value.snapshot.projection.visibleTurnItems.pop();
+          },
+          (value: typeof partial) => {
+            value.snapshot.projection.visibleTurnItems.shift();
+          },
+          (value: typeof partial) => {
+            value.snapshot.projection.turnItems.pop();
+          },
+          (value: typeof partial) => {
+            value.snapshot.turnItemCount = 0;
+          },
+          (value: typeof partial) => {
+            value.snapshot.projection.visibleTurnItems[1].sourceItemId = "wrong-item";
+          },
+          (value: typeof partial) => {
+            value.snapshot.projection.visibleTurnItems[1].visibility = "inherited";
+          },
+          (value: typeof partial) => {
+            value.snapshot.projection.visibleTurnItems[1].item.threadId = "wrong-thread";
+          },
+          (value: typeof partial) => {
+            value.snapshot.projection.thread.id = "wrong-thread";
+          },
+          (value: typeof partial) => {
+            value.snapshot.projection.turnItems.push(
+              value.snapshot.projection.visibleTurnItems[1].item,
+            );
+          },
+          (value: typeof partial) => {
+            value.snapshot.projection.visibleTurnItems.push(
+              value.snapshot.projection.visibleTurnItems[1],
+            );
+          },
+        ].map((corrupt) => {
+          const value = JSON.parse(raw);
+          corrupt(value);
+          return value;
+        }),
+      ];
+      const invalidRecords = ["{broken", ...corruptions.map((value) => JSON.stringify(value))];
+      for (const invalid of invalidRecords) {
+        memory.values.set(id, invalid);
+        expect(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)).toEqual(Option.none());
+        expect(memory.values.has(id)).toBe(false);
+      }
+      expect(memory.removed).toEqual(invalidRecords.map(() => id));
+      yield* store.saveThread(ENVIRONMENT_ID, threadSnapshotWithItems());
+      expect(Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID))).toEqual(
+        threadSnapshotWithItems(),
+      );
+    }),
+  );
+
+  it.effect("round-trips V5 exactly and restores shared items before returning to a consumer", () =>
+    Effect.gen(function* () {
+      const memory = makeDatabase();
+      const store = yield* make().pipe(
+        Effect.provideService(MobileDatabase.MobileDatabase, memory.database),
+      );
+      const original = threadSnapshotWithItems();
+      const local = original.projection.visibleTurnItems[1]!;
+      const inherited = original.projection.visibleTurnItems[0]!;
+      const synthetic = { ...local.item, id: TurnItemId.make("synthetic-notice"), ordinal: 12 };
+      const fork = {
+        id: TurnItemId.make("fork-marker"),
+        threadId: THREAD_ID,
+        runId: null,
+        nodeId: null,
+        providerTurnId: null,
+        nativeItemRef: null,
+        parentItemId: null,
+        ordinal: 0,
+        status: "completed" as const,
+        title: null,
+        startedAt: null,
+        completedAt: NOW,
+        updatedAt: NOW,
+        type: "fork" as const,
+        source: {
+          type: "run" as const,
+          threadId: ThreadId.make("parent-thread"),
+          runId: RunId.make("parent-run"),
+        },
+        targetThreadId: THREAD_ID,
+      };
+      const snapshot = {
+        ...original,
+        projection: {
+          ...original.projection,
+          turnItems: [local.item, synthetic, original.projection.turnItems[1]!],
+          visibleTurnItems: [
+            { ...local, position: 0 },
+            {
+              ...inherited,
+              position: 1,
+              sourceItemId: local.sourceItemId,
+              item: { ...inherited.item, id: local.item.id },
+            },
+            {
+              ...local,
+              position: 2,
+              visibility: "synthetic" as const,
+              sourceItemId: synthetic.id,
+              item: synthetic,
+            },
+            {
+              position: 3,
+              visibility: "synthetic" as const,
+              sourceThreadId: fork.source.threadId,
+              sourceItemId: fork.id,
+              item: fork,
+            },
+          ],
+        },
+      };
+      yield* store.saveThread(ENVIRONMENT_ID, snapshot);
+      const id = cacheId(ENVIRONMENT_ID, "thread", THREAD_ID);
+      const encoded = JSON.parse(memory.values.get(id)!);
+      expect(encoded.schemaVersion).toBe(6);
+      expect(memory.schemaVersions.get(id)).toBe(6);
+      expect(encoded.snapshot.turnItemsOmitLocalVisible).toBe(true);
+      expect(encoded.snapshot.turnItemCount).toBe(3);
+      expect(encoded.snapshot.visibleTurnItemCount).toBe(4);
+      expect(encoded.snapshot.projection.turnItems.map((item: { id: string }) => item.id)).toEqual([
+        "hidden-notice",
+      ]);
+
+      const duplicate = JSON.parse(memory.values.get(id)!);
+      duplicate.snapshot.projection.visibleTurnItems.push({
+        ...duplicate.snapshot.projection.visibleTurnItems[1],
+        position: 4,
+      });
+      duplicate.snapshot.visibleTurnItemCount += 1;
+      memory.values.set(id, JSON.stringify(duplicate));
+      expect(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)).toEqual(Option.none());
+      expect(memory.values.has(id)).toBe(false);
+      expect(memory.removed).toEqual([id]);
+      yield* store.saveThread(ENVIRONMENT_ID, snapshot);
+
+      const loaded = Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID));
+      expect(loaded).toEqual(snapshot);
+      expect(loaded).not.toHaveProperty("turnItemsOmitLocalVisible");
+      expect(loaded).not.toHaveProperty("turnItemCount");
+      expect(loaded).not.toHaveProperty("visibleTurnItemCount");
+      expect(loaded.projection.turnItems[0]).toBe(loaded.projection.visibleTurnItems[0]!.item);
+      expect(loaded.projection.turnItems[1]).toBe(loaded.projection.visibleTurnItems[2]!.item);
+      expect(snapshot.projection.turnItems).toHaveLength(3);
+    }),
+  );
+
+  it.effect("keeps a full projection when visible rows cannot reproduce the original items", () =>
+    Effect.gen(function* () {
+      const memory = makeDatabase();
+      const store = yield* make().pipe(
+        Effect.provideService(MobileDatabase.MobileDatabase, memory.database),
+      );
+      const original = threadSnapshotWithItems();
+      const local = original.projection.turnItems[0]!;
+      for (const turnItems of [
+        [{ ...local }, original.projection.turnItems[1]!],
+        [{ ...local, message: "Newer than the visible row." }, original.projection.turnItems[1]!],
+        original.projection.turnItems.toReversed(),
+      ]) {
+        const snapshot = { ...original, projection: { ...original.projection, turnItems } };
+        yield* store.saveThread(ENVIRONMENT_ID, snapshot);
+        const raw = memory.values.get(cacheId(ENVIRONMENT_ID, "thread", THREAD_ID))!;
+        const encoded = JSON.parse(raw);
+        expect(encoded.schemaVersion).toBe(6);
+        expect(encoded.snapshot.turnItemsOmitLocalVisible).toBe(false);
+        expect(encoded.snapshot.turnItemCount).toBe(2);
+        expect(encoded.snapshot.projection.turnItems).toHaveLength(2);
+        expect(Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID))).toEqual(
+          snapshot,
+        );
+      }
+    }),
+  );
+
   it.effect("round-trips V2 shell and thread DateTime fields with the shared cache schema", () =>
     Effect.gen(function* () {
       const memory = makeDatabase();
@@ -229,7 +522,7 @@ describe("mobile SQLite environment cache store", () => {
         ORCHESTRATION_CACHE_SCHEMA_VERSION,
       );
       expect(memory.schemaVersions.get(cacheId(ENVIRONMENT_ID, "thread", THREAD_ID))).toBe(
-        ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
+        COMPACT_THREAD_CACHE_SCHEMA_VERSION,
       );
     }),
   );

apps/mobile/src/connection/environment-cache-store.ts

@@ -1,7 +1,8 @@
 import {
   ORCHESTRATION_CACHE_SCHEMA_VERSION,
-  ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
-  StoredOrchestrationThreadSnapshot,
+  COMPACT_THREAD_CACHE_SCHEMA_VERSION,
+  decodeStoredThreadSnapshot,
+  encodeStoredThreadSnapshot,
   Persistence,
   makeStoredShellSnapshotEncoder,
 } from "@t3tools/client-runtime/platform";
@@ -33,12 +34,6 @@ const StoredVcsRefs = Schema.Struct({
   refs: VcsListRefsResult,
 });
 
-const decodeStoredThreadSnapshot = Schema.decodeUnknownEffect(
-  Schema.fromJsonString(StoredOrchestrationThreadSnapshot),
-);
-const encodeStoredThreadSnapshot = Schema.encodeEffect(
-  Schema.fromJsonString(StoredOrchestrationThreadSnapshot),
-);
 const decodeStoredServerConfig = Schema.decodeUnknownEffect(
   Schema.fromJsonString(StoredServerConfig),
 );
@@ -147,19 +142,12 @@ 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_THREAD_CACHE_SCHEMA_VERSION,
         environmentId,
         threadId,
         snapshot,
       }).pipe(Effect.mapError((cause) => persistenceError("save-thread", cause)));
       yield* database
-        .saveCache(
-          environmentId,
-          "thread",
-          threadId,
-          ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
-          payload,
-        )
+        .saveCache(environmentId, "thread", threadId, COMPACT_THREAD_CACHE_SCHEMA_VERSION, payload)
         .pipe(Effect.mapError(mapDatabaseError("save-thread")));
     }),
     removeThread: Effect.fn("MobileEnvironmentCache.removeThread")((environmentId, threadId) =>

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

@@ -4,5 +4,6 @@ export * as Persistence from "./persistence.ts";
 export { ConnectionPersistenceError } from "./persistence.ts";
 export * from "./orchestrationCache.ts";
 export * from "./shellCacheEncoding.ts";
+export * from "./threadCacheEncoding.ts";
 export * as PlatformConnectionSource from "./source.ts";
 export * from "./storageDocument.ts";