From 71e20db6ecd3d91ee4c0cef44cd29d9cd90748d2 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 23 Oct 2026 16:32:37 -0700 Subject: [PATCH 1/5] perf(mobile): compact persisted thread snapshots --- .../environment-cache-store.test.ts | 135 ++++++++++++- .../src/connection/environment-cache-store.ts | 20 +- packages/client-runtime/src/platform/index.ts | 1 + .../src/platform/threadCacheEncoding.test.ts | 191 ++++++++++++++++++ .../src/platform/threadCacheEncoding.ts | 58 ++++++ 5 files changed, 388 insertions(+), 17 deletions(-) create mode 100644 packages/client-runtime/src/platform/threadCacheEncoding.test.ts create mode 100644 packages/client-runtime/src/platform/threadCacheEncoding.ts diff --git a/apps/mobile/src/connection/environment-cache-store.test.ts b/apps/mobile/src/connection/environment-cache-store.test.ts index 4c88a03ccf2a..b2a2b62d2f9d 100644 --- a/apps/mobile/src/connection/environment-cache-store.test.ts +++ b/apps/mobile/src/connection/environment-cache-store.test.ts @@ -1,6 +1,8 @@ import { ORCHESTRATION_CACHE_SCHEMA_VERSION, ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, + COMPACT_THREAD_CACHE_SCHEMA_VERSION, + StoredOrchestrationThreadSnapshot, } from "@t3tools/client-runtime/platform"; import { CommandId, @@ -8,6 +10,7 @@ import { ProjectId, ProviderInstanceId, ThreadId, + TurnItemId, type OrchestrationV2ShellSnapshot, type OrchestrationV2ThreadDetailSnapshot, type ThreadPullRequestLink, @@ -17,6 +20,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 +30,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 +159,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(); const schemaVersions = new Map(); @@ -198,6 +258,79 @@ function makeDatabase() { } describe("mobile SQLite environment cache store", () => { + it.effect("loads a V4 record and migrates to V5 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 obsolete and corrupt thread records and rejects another environment's cache", + () => + 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)!; + for (const invalid of ["{broken", raw.replace('"schemaVersion":5', '"schemaVersion":3')]) { + memory.values.set(id, invalid); + expect(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)).toEqual(Option.none()); + } + expect(memory.removed).toEqual([id, id]); + memory.values.set( + id, + raw.replace('"environmentId":"environment-1"', '"environmentId":"other"'), + ); + expect(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)).toEqual(Option.none()); + memory.values.set(id, raw.replace('"threadId":"thread-1"', '"threadId":"other"')); + expect(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)).toEqual(Option.none()); + }), + ); + + it.effect( + "persists each local timeline payload once and restores shared items before returning", + () => + Effect.gen(function* () { + const memory = makeDatabase(); + const store = yield* make().pipe( + Effect.provideService(MobileDatabase.MobileDatabase, memory.database), + ); + const snapshot = threadSnapshotWithItems(); + yield* store.saveThread(ENVIRONMENT_ID, snapshot); + const raw = memory.values.get(cacheId(ENVIRONMENT_ID, "thread", THREAD_ID))!; + expect(raw.match(/"id":"local-notice"/g)).toHaveLength(1); + + const loaded = Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)); + expect(loaded).toEqual(snapshot); + expect(loaded.projection.turnItems[0]).toBe(loaded.projection.visibleTurnItems[1]!.item); + expect(snapshot.projection.turnItems).toHaveLength(2); + }), + ); + it.effect("round-trips V2 shell and thread DateTime fields with the shared cache schema", () => Effect.gen(function* () { const memory = makeDatabase(); @@ -229,7 +362,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, ); }), ); diff --git a/apps/mobile/src/connection/environment-cache-store.ts b/apps/mobile/src/connection/environment-cache-store.ts index 8df7fc104df2..b55af4718d15 100644 --- a/apps/mobile/src/connection/environment-cache-store.ts +++ b/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) => diff --git a/packages/client-runtime/src/platform/index.ts b/packages/client-runtime/src/platform/index.ts index 18efb9be18bc..fa7be91aecd3 100644 --- a/packages/client-runtime/src/platform/index.ts +++ b/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"; diff --git a/packages/client-runtime/src/platform/threadCacheEncoding.test.ts b/packages/client-runtime/src/platform/threadCacheEncoding.test.ts new file mode 100644 index 000000000000..88152152106c --- /dev/null +++ b/packages/client-runtime/src/platform/threadCacheEncoding.test.ts @@ -0,0 +1,191 @@ +import { describe, expect, it } from "@effect/vitest"; +import { EnvironmentId, ThreadId, TurnItemId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Result from "effect/Result"; +import * as Schema from "effect/Schema"; + +import { + ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, + StoredOrchestrationThreadSnapshot, +} from "./orchestrationCache.ts"; +import { + COMPACT_THREAD_CACHE_SCHEMA_VERSION, + decodeStoredThreadSnapshot, + encodeStoredThreadSnapshot, +} from "./threadCacheEncoding.ts"; +import { v2Now, v2Projection, v2ThreadId } from "../state/orchestrationV2TestFixtures.ts"; + +const encodeLegacy = Schema.encodeSync(Schema.fromJsonString(StoredOrchestrationThreadSnapshot)); +const decodeLegacy = Schema.decodeUnknownSync( + Schema.fromJsonString(StoredOrchestrationThreadSnapshot), +); + +function fixture() { + const item = { + id: TurnItemId.make("local-cache-item"), + threadId: v2ThreadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 12, + status: "completed" as const, + title: null, + startedAt: v2Now, + completedAt: v2Now, + updatedAt: v2Now, + type: "system_notice" as const, + message: "Keep the complete notice.", + }; + const hidden = { ...item, id: TurnItemId.make("hidden-cache-item"), ordinal: 13 }; + const inherited = { + ...item, + id: TurnItemId.make("parent-cache-item"), + threadId: ThreadId.make("parent-thread"), + }; + return { + schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, + environmentId: EnvironmentId.make("cache-environment"), + threadId: v2ThreadId, + snapshot: { + snapshotSequence: 42, + historyCursor: "older-activity", + hasMoreHistory: true, + latestLocalTurnOrdinal: hidden.ordinal, + projection: { + ...v2Projection, + turnItems: [item, hidden], + visibleTurnItems: [ + { + position: 0, + visibility: "inherited" as const, + sourceThreadId: inherited.threadId, + sourceItemId: inherited.id, + item: inherited, + }, + { + position: 1, + visibility: "local" as const, + sourceThreadId: v2ThreadId, + sourceItemId: item.id, + item, + }, + ], + }, + }, + }; +} + +describe("compact thread cache encoding", () => { + it.effect( + "restores local identity, order, hidden items, inherited rows, dates, and paging metadata", + () => + Effect.gen(function* () { + const stored = fixture(); + const raw = yield* encodeStoredThreadSnapshot(stored); + const encoded = JSON.parse(raw); + expect(encoded.schemaVersion).toBe(COMPACT_THREAD_CACHE_SCHEMA_VERSION); + expect(encoded.snapshot.turnItemsOmitLocalVisible).toBe(true); + expect( + encoded.snapshot.projection.turnItems.map((item: { id: string }) => item.id), + ).toEqual(["hidden-cache-item"]); + const decoded = yield* decodeStoredThreadSnapshot(raw); + expect(decoded.snapshot).toEqual(stored.snapshot); + expect(decoded.snapshot).not.toHaveProperty("turnItemsOmitLocalVisible"); + expect(decoded.snapshot.projection.turnItems[0]).toBe( + decoded.snapshot.projection.visibleTurnItems[1]!.item, + ); + expect(yield* encodeStoredThreadSnapshot(decoded)).toBe(raw); + expect(() => decodeLegacy(raw)).toThrow(); + expect(stored.snapshot.projection.turnItems).toHaveLength(2); + }), + ); + + it.effect("reads V4 without restoring an unmarked full projection, then saves as V5", () => + Effect.gen(function* () { + const stored = fixture(); + const decoded = yield* decodeStoredThreadSnapshot(encodeLegacy(stored)); + expect(decoded).toEqual(stored); + // V4 JSON lost object sharing. Compaction must conservatively keep those entries. + const migrated = yield* encodeStoredThreadSnapshot(decoded); + expect(JSON.parse(migrated).schemaVersion).toBe(COMPACT_THREAD_CACHE_SCHEMA_VERSION); + expect(JSON.parse(migrated).snapshot).not.toHaveProperty("turnItemsOmitLocalVisible"); + expect((yield* decodeStoredThreadSnapshot(migrated)).snapshot).toEqual(stored.snapshot); + }), + ); + + it.effect.each(["different-value", "different-object", "interleaved-hidden", "empty"] as const)( + "keeps the full projection when omission is unsafe or unnecessary: %s", + (scenario) => + Effect.gen(function* () { + const stored = fixture(); + const projection = stored.snapshot.projection; + const local = projection.turnItems[0]!; + if (scenario === "different-value") { + projection.turnItems[0] = { ...local, message: "A different version of the same item." }; + } else if (scenario === "different-object") { + projection.turnItems[0] = { ...local }; + } else if (scenario === "interleaved-hidden") { + projection.turnItems.reverse(); + } else { + projection.turnItems = []; + projection.visibleTurnItems = []; + } + const raw = yield* encodeStoredThreadSnapshot(stored); + expect(JSON.parse(raw).snapshot).not.toHaveProperty("turnItemsOmitLocalVisible"); + expect((yield* decodeStoredThreadSnapshot(raw)).snapshot).toEqual(stored.snapshot); + }), + ); + + it.effect("restores multiple locals around inherited and same-thread synthetic rows", () => + Effect.gen(function* () { + const stored = fixture(); + const projection = stored.snapshot.projection; + const localRow = projection.visibleTurnItems[1]!; + const second = { ...localRow.item, id: TurnItemId.make("second-local"), ordinal: 14 }; + const snapshot = { + ...stored.snapshot, + projection: { + ...projection, + turnItems: [localRow.item, second, projection.turnItems[1]!], + visibleTurnItems: [ + { ...localRow, position: 0 }, + { ...projection.visibleTurnItems[0]!, position: 1 }, + { + ...localRow, + position: 2, + visibility: "synthetic" as const, + sourceItemId: second.id, + item: second, + }, + ], + }, + }; + const decoded = yield* decodeStoredThreadSnapshot( + yield* encodeStoredThreadSnapshot({ ...stored, snapshot }), + ); + expect(decoded.snapshot).toEqual(snapshot); + expect(decoded.snapshot.projection.turnItems[1]).toBe( + decoded.snapshot.projection.visibleTurnItems[2]!.item, + ); + }), + ); + + it.effect("rejects invalid markers, invalid items, obsolete V3, and future versions", () => + Effect.gen(function* () { + const raw = yield* encodeStoredThreadSnapshot(fixture()); + for (const invalid of [ + raw.replace('"turnItemsOmitLocalVisible":true', '"turnItemsOmitLocalVisible":false'), + raw.replace('"position":1', '"position":-1'), + raw.replace('"schemaVersion":5', '"schemaVersion":3'), + raw.replace('"schemaVersion":5', '"schemaVersion":6'), + ]) { + expect(Result.isFailure(yield* Effect.result(decodeStoredThreadSnapshot(invalid)))).toBe( + true, + ); + } + }), + ); +}); diff --git a/packages/client-runtime/src/platform/threadCacheEncoding.ts b/packages/client-runtime/src/platform/threadCacheEncoding.ts new file mode 100644 index 000000000000..4a89edde5de1 --- /dev/null +++ b/packages/client-runtime/src/platform/threadCacheEncoding.ts @@ -0,0 +1,58 @@ +import { + boundedSnapshotProjection, + omitLocalVisibleTurnItems, +} from "@t3tools/shared/orchestrationV2BoundedSnapshot"; +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; + +import { StoredOrchestrationThreadSnapshot } from "./orchestrationCache.ts"; + +// Keep the full V4 envelope for web/desktop and for reading existing mobile caches. +// Older clients must discard V5 rather than use a projection with missing items. +export const COMPACT_THREAD_CACHE_SCHEMA_VERSION = 5 as const; + +const StoredCompactThreadSnapshot = StoredOrchestrationThreadSnapshot.mapFields((fields) => ({ + ...fields, + schemaVersion: Schema.Literal(COMPACT_THREAD_CACHE_SCHEMA_VERSION), + snapshot: fields.snapshot.mapFields((snapshotFields) => ({ + ...snapshotFields, + turnItemsOmitLocalVisible: Schema.optionalKey(Schema.Literal(true)), + })), +})); + +const encodeCompact = Schema.encodeEffect(Schema.fromJsonString(StoredCompactThreadSnapshot)); +const decodeCompatible = Schema.decodeUnknownEffect( + Schema.fromJsonString( + Schema.Union([StoredCompactThreadSnapshot, StoredOrchestrationThreadSnapshot]), + ), +); + +export const encodeStoredThreadSnapshot = Effect.fn("ThreadCache.encode")(function* ( + stored: Pick< + typeof StoredOrchestrationThreadSnapshot.Type, + "environmentId" | "threadId" | "snapshot" + >, +) { + const { snapshot } = stored; + const compact = omitLocalVisibleTurnItems(snapshot.projection); + return yield* encodeCompact({ + ...stored, + schemaVersion: COMPACT_THREAD_CACHE_SCHEMA_VERSION, + snapshot: compact + ? { ...snapshot, projection: compact, turnItemsOmitLocalVisible: true } + : snapshot, + }); +}); + +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; + const { turnItemsOmitLocalVisible, ...snapshot } = stored.snapshot; + return { + ...stored, + snapshot: { + ...snapshot, + projection: boundedSnapshotProjection({ ...snapshot, turnItemsOmitLocalVisible }), + }, + }; +}); From 1c9012e81a1ed8ef07501aa4774182310165809c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 03:51:57 +0000 Subject: [PATCH 2/5] test(mobile): focus thread cache roundtrip and recovery coverage --- .../environment-cache-store.test.ts | 131 ++++++++---- .../src/platform/threadCacheEncoding.test.ts | 191 ------------------ .../src/state/threads-sync.test.ts | 43 ++-- 3 files changed, 116 insertions(+), 249 deletions(-) delete mode 100644 packages/client-runtime/src/platform/threadCacheEncoding.test.ts diff --git a/apps/mobile/src/connection/environment-cache-store.test.ts b/apps/mobile/src/connection/environment-cache-store.test.ts index b2a2b62d2f9d..116b1fb4b327 100644 --- a/apps/mobile/src/connection/environment-cache-store.test.ts +++ b/apps/mobile/src/connection/environment-cache-store.test.ts @@ -285,50 +285,103 @@ describe("mobile SQLite environment cache store", () => { }), ); - it.effect( - "discards obsolete and corrupt thread records and rejects another environment's cache", - () => - 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)!; - for (const invalid of ["{broken", raw.replace('"schemaVersion":5', '"schemaVersion":3')]) { - memory.values.set(id, invalid); - expect(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)).toEqual(Option.none()); - } - expect(memory.removed).toEqual([id, id]); - memory.values.set( - id, - raw.replace('"environmentId":"environment-1"', '"environmentId":"other"'), - ); - expect(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)).toEqual(Option.none()); - memory.values.set(id, raw.replace('"threadId":"thread-1"', '"threadId":"other"')); + 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; + for (const invalid of ["{broken", JSON.stringify(partial), JSON.stringify(invalidItem)]) { + 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([id, id, id]); + yield* store.saveThread(ENVIRONMENT_ID, threadSnapshotWithItems()); + expect(Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID))).toEqual( + threadSnapshotWithItems(), + ); + }), ); - it.effect( - "persists each local timeline payload once and restores shared items before returning", - () => - Effect.gen(function* () { - const memory = makeDatabase(); - const store = yield* make().pipe( - Effect.provideService(MobileDatabase.MobileDatabase, memory.database), - ); - const snapshot = 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 synthetic = { ...local.item, id: TurnItemId.make("synthetic-notice"), ordinal: 12 }; + const snapshot = { + ...original, + projection: { + ...original.projection, + turnItems: [local.item, synthetic, original.projection.turnItems[1]!], + visibleTurnItems: [ + { ...local, position: 0 }, + { ...original.projection.visibleTurnItems[0]!, position: 1 }, + { + ...local, + position: 2, + visibility: "synthetic" as const, + sourceItemId: synthetic.id, + item: synthetic, + }, + ], + }, + }; + 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(5); + expect(memory.schemaVersions.get(id)).toBe(5); + expect(encoded.snapshot.turnItemsOmitLocalVisible).toBe(true); + expect(encoded.snapshot.projection.turnItems.map((item: { id: string }) => item.id)).toEqual([ + "hidden-notice", + ]); + + const loaded = Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)); + expect(loaded).toEqual(snapshot); + expect(loaded).not.toHaveProperty("turnItemsOmitLocalVisible"); + 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))!; - expect(raw.match(/"id":"local-notice"/g)).toHaveLength(1); - - const loaded = Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)); - expect(loaded).toEqual(snapshot); - expect(loaded.projection.turnItems[0]).toBe(loaded.projection.visibleTurnItems[1]!.item); - expect(snapshot.projection.turnItems).toHaveLength(2); - }), + const encoded = JSON.parse(raw); + expect(encoded.schemaVersion).toBe(5); + expect(encoded.snapshot).not.toHaveProperty("turnItemsOmitLocalVisible"); + 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", () => diff --git a/packages/client-runtime/src/platform/threadCacheEncoding.test.ts b/packages/client-runtime/src/platform/threadCacheEncoding.test.ts deleted file mode 100644 index 88152152106c..000000000000 --- a/packages/client-runtime/src/platform/threadCacheEncoding.test.ts +++ /dev/null @@ -1,191 +0,0 @@ -import { describe, expect, it } from "@effect/vitest"; -import { EnvironmentId, ThreadId, TurnItemId } from "@t3tools/contracts"; -import * as Effect from "effect/Effect"; -import * as Result from "effect/Result"; -import * as Schema from "effect/Schema"; - -import { - ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, - StoredOrchestrationThreadSnapshot, -} from "./orchestrationCache.ts"; -import { - COMPACT_THREAD_CACHE_SCHEMA_VERSION, - decodeStoredThreadSnapshot, - encodeStoredThreadSnapshot, -} from "./threadCacheEncoding.ts"; -import { v2Now, v2Projection, v2ThreadId } from "../state/orchestrationV2TestFixtures.ts"; - -const encodeLegacy = Schema.encodeSync(Schema.fromJsonString(StoredOrchestrationThreadSnapshot)); -const decodeLegacy = Schema.decodeUnknownSync( - Schema.fromJsonString(StoredOrchestrationThreadSnapshot), -); - -function fixture() { - const item = { - id: TurnItemId.make("local-cache-item"), - threadId: v2ThreadId, - runId: null, - nodeId: null, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: 12, - status: "completed" as const, - title: null, - startedAt: v2Now, - completedAt: v2Now, - updatedAt: v2Now, - type: "system_notice" as const, - message: "Keep the complete notice.", - }; - const hidden = { ...item, id: TurnItemId.make("hidden-cache-item"), ordinal: 13 }; - const inherited = { - ...item, - id: TurnItemId.make("parent-cache-item"), - threadId: ThreadId.make("parent-thread"), - }; - return { - schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, - environmentId: EnvironmentId.make("cache-environment"), - threadId: v2ThreadId, - snapshot: { - snapshotSequence: 42, - historyCursor: "older-activity", - hasMoreHistory: true, - latestLocalTurnOrdinal: hidden.ordinal, - projection: { - ...v2Projection, - turnItems: [item, hidden], - visibleTurnItems: [ - { - position: 0, - visibility: "inherited" as const, - sourceThreadId: inherited.threadId, - sourceItemId: inherited.id, - item: inherited, - }, - { - position: 1, - visibility: "local" as const, - sourceThreadId: v2ThreadId, - sourceItemId: item.id, - item, - }, - ], - }, - }, - }; -} - -describe("compact thread cache encoding", () => { - it.effect( - "restores local identity, order, hidden items, inherited rows, dates, and paging metadata", - () => - Effect.gen(function* () { - const stored = fixture(); - const raw = yield* encodeStoredThreadSnapshot(stored); - const encoded = JSON.parse(raw); - expect(encoded.schemaVersion).toBe(COMPACT_THREAD_CACHE_SCHEMA_VERSION); - expect(encoded.snapshot.turnItemsOmitLocalVisible).toBe(true); - expect( - encoded.snapshot.projection.turnItems.map((item: { id: string }) => item.id), - ).toEqual(["hidden-cache-item"]); - const decoded = yield* decodeStoredThreadSnapshot(raw); - expect(decoded.snapshot).toEqual(stored.snapshot); - expect(decoded.snapshot).not.toHaveProperty("turnItemsOmitLocalVisible"); - expect(decoded.snapshot.projection.turnItems[0]).toBe( - decoded.snapshot.projection.visibleTurnItems[1]!.item, - ); - expect(yield* encodeStoredThreadSnapshot(decoded)).toBe(raw); - expect(() => decodeLegacy(raw)).toThrow(); - expect(stored.snapshot.projection.turnItems).toHaveLength(2); - }), - ); - - it.effect("reads V4 without restoring an unmarked full projection, then saves as V5", () => - Effect.gen(function* () { - const stored = fixture(); - const decoded = yield* decodeStoredThreadSnapshot(encodeLegacy(stored)); - expect(decoded).toEqual(stored); - // V4 JSON lost object sharing. Compaction must conservatively keep those entries. - const migrated = yield* encodeStoredThreadSnapshot(decoded); - expect(JSON.parse(migrated).schemaVersion).toBe(COMPACT_THREAD_CACHE_SCHEMA_VERSION); - expect(JSON.parse(migrated).snapshot).not.toHaveProperty("turnItemsOmitLocalVisible"); - expect((yield* decodeStoredThreadSnapshot(migrated)).snapshot).toEqual(stored.snapshot); - }), - ); - - it.effect.each(["different-value", "different-object", "interleaved-hidden", "empty"] as const)( - "keeps the full projection when omission is unsafe or unnecessary: %s", - (scenario) => - Effect.gen(function* () { - const stored = fixture(); - const projection = stored.snapshot.projection; - const local = projection.turnItems[0]!; - if (scenario === "different-value") { - projection.turnItems[0] = { ...local, message: "A different version of the same item." }; - } else if (scenario === "different-object") { - projection.turnItems[0] = { ...local }; - } else if (scenario === "interleaved-hidden") { - projection.turnItems.reverse(); - } else { - projection.turnItems = []; - projection.visibleTurnItems = []; - } - const raw = yield* encodeStoredThreadSnapshot(stored); - expect(JSON.parse(raw).snapshot).not.toHaveProperty("turnItemsOmitLocalVisible"); - expect((yield* decodeStoredThreadSnapshot(raw)).snapshot).toEqual(stored.snapshot); - }), - ); - - it.effect("restores multiple locals around inherited and same-thread synthetic rows", () => - Effect.gen(function* () { - const stored = fixture(); - const projection = stored.snapshot.projection; - const localRow = projection.visibleTurnItems[1]!; - const second = { ...localRow.item, id: TurnItemId.make("second-local"), ordinal: 14 }; - const snapshot = { - ...stored.snapshot, - projection: { - ...projection, - turnItems: [localRow.item, second, projection.turnItems[1]!], - visibleTurnItems: [ - { ...localRow, position: 0 }, - { ...projection.visibleTurnItems[0]!, position: 1 }, - { - ...localRow, - position: 2, - visibility: "synthetic" as const, - sourceItemId: second.id, - item: second, - }, - ], - }, - }; - const decoded = yield* decodeStoredThreadSnapshot( - yield* encodeStoredThreadSnapshot({ ...stored, snapshot }), - ); - expect(decoded.snapshot).toEqual(snapshot); - expect(decoded.snapshot.projection.turnItems[1]).toBe( - decoded.snapshot.projection.visibleTurnItems[2]!.item, - ); - }), - ); - - it.effect("rejects invalid markers, invalid items, obsolete V3, and future versions", () => - Effect.gen(function* () { - const raw = yield* encodeStoredThreadSnapshot(fixture()); - for (const invalid of [ - raw.replace('"turnItemsOmitLocalVisible":true', '"turnItemsOmitLocalVisible":false'), - raw.replace('"position":1', '"position":-1'), - raw.replace('"schemaVersion":5', '"schemaVersion":3'), - raw.replace('"schemaVersion":5', '"schemaVersion":6'), - ]) { - expect(Result.isFailure(yield* Effect.result(decodeStoredThreadSnapshot(invalid)))).toBe( - true, - ); - } - }), - ); -}); diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index b87ab18a50ae..8ef8c14b1b28 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -320,26 +320,31 @@ const deleted = (sequence = 3): OrchestrationV2ThreadStreamItem => { }; describe("EnvironmentThreads", () => { - it.effect("loads the server thread when its local cache read fails", () => - Effect.gen(function* () { - const h = yield* makeHarness({ - loadCached: Effect.fail( - new Persistence.ConnectionPersistenceError({ - operation: "load-thread", - message: "The database connection is closing.", - }), - ), - httpSnapshot: { - _tag: "present", - snapshot: { snapshotSequence: 7, projection: BASE_PROJECTION }, - }, - }); + it.effect.each(["read failure", "cache miss"] as const)( + "loads the server thread after a %s", + (scenario) => + Effect.gen(function* () { + const h = yield* makeHarness({ + loadCached: + scenario === "cache miss" + ? Effect.succeedNone + : Effect.fail( + new Persistence.ConnectionPersistenceError({ + operation: "load-thread", + message: "The database connection is closing.", + }), + ), + httpSnapshot: { + _tag: "present", + snapshot: { snapshotSequence: 7, projection: BASE_PROJECTION }, + }, + }); - const state = yield* awaitThreadState(h.observed, (value) => value.status === "live"); - expect(Option.getOrThrow(state.data)).toEqual(BASE_PROJECTION); - expect(yield* Ref.get(h.loaderCalls)).toBe(1); - expect(yield* Ref.get(h.subscriptionCount)).toBe(1); - }), + const state = yield* awaitThreadState(h.observed, (value) => value.status === "live"); + expect(Option.getOrThrow(state.data)).toEqual(BASE_PROJECTION); + expect(yield* Ref.get(h.loaderCalls)).toBe(1); + expect(yield* Ref.get(h.subscriptionCount)).toBe(1); + }), ); it.effect("flushes newer data after an older background write completes", () => From 18dace5064692fc8f6613b32b442f6b2c34af5a6 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 03:57:50 +0000 Subject: [PATCH 3/5] fix(mobile): reject inconsistent compact thread caches --- .../environment-cache-store.test.ts | 95 ++++++++++++++++++- .../src/platform/threadCacheEncoding.ts | 74 ++++++++++++--- 2 files changed, 151 insertions(+), 18 deletions(-) diff --git a/apps/mobile/src/connection/environment-cache-store.test.ts b/apps/mobile/src/connection/environment-cache-store.test.ts index 116b1fb4b327..6857bc56ec88 100644 --- a/apps/mobile/src/connection/environment-cache-store.test.ts +++ b/apps/mobile/src/connection/environment-cache-store.test.ts @@ -9,6 +9,7 @@ import { EnvironmentId, ProjectId, ProviderInstanceId, + RunId, ThreadId, TurnItemId, type OrchestrationV2ShellSnapshot, @@ -298,12 +299,66 @@ describe("mobile SQLite environment cache store", () => { delete partial.snapshot.projection.visibleTurnItems; const invalidItem = JSON.parse(raw); delete invalidItem.snapshot.projection.visibleTurnItems[1].item.message; - for (const invalid of ["{broken", JSON.stringify(partial), JSON.stringify(invalidItem)]) { + 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([id, id, id]); + 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(), @@ -320,6 +375,28 @@ describe("mobile SQLite environment cache store", () => { const original = threadSnapshotWithItems(); const local = original.projection.visibleTurnItems[1]!; 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: { @@ -335,6 +412,13 @@ describe("mobile SQLite environment cache store", () => { sourceItemId: synthetic.id, item: synthetic, }, + { + position: 3, + visibility: "synthetic" as const, + sourceThreadId: fork.source.threadId, + sourceItemId: fork.id, + item: fork, + }, ], }, }; @@ -344,6 +428,8 @@ describe("mobile SQLite environment cache store", () => { expect(encoded.schemaVersion).toBe(5); expect(memory.schemaVersions.get(id)).toBe(5); 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", ]); @@ -351,6 +437,8 @@ describe("mobile SQLite environment cache store", () => { 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); @@ -375,7 +463,8 @@ describe("mobile SQLite environment cache store", () => { const raw = memory.values.get(cacheId(ENVIRONMENT_ID, "thread", THREAD_ID))!; const encoded = JSON.parse(raw); expect(encoded.schemaVersion).toBe(5); - expect(encoded.snapshot).not.toHaveProperty("turnItemsOmitLocalVisible"); + 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, diff --git a/packages/client-runtime/src/platform/threadCacheEncoding.ts b/packages/client-runtime/src/platform/threadCacheEncoding.ts index 4a89edde5de1..ad5ac74cc412 100644 --- a/packages/client-runtime/src/platform/threadCacheEncoding.ts +++ b/packages/client-runtime/src/platform/threadCacheEncoding.ts @@ -1,14 +1,14 @@ import { - boundedSnapshotProjection, omitLocalVisibleTurnItems, + restoreLocalVisibleTurnItems, } from "@t3tools/shared/orchestrationV2BoundedSnapshot"; +import * as Data from "effect/Data"; import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; import { StoredOrchestrationThreadSnapshot } from "./orchestrationCache.ts"; -// Keep the full V4 envelope for web/desktop and for reading existing mobile caches. -// Older clients must discard V5 rather than use a projection with missing items. +// Keep V4 for web/desktop and existing mobile caches; older clients discard V5. export const COMPACT_THREAD_CACHE_SCHEMA_VERSION = 5 as const; const StoredCompactThreadSnapshot = StoredOrchestrationThreadSnapshot.mapFields((fields) => ({ @@ -16,10 +16,15 @@ const StoredCompactThreadSnapshot = StoredOrchestrationThreadSnapshot.mapFields( schemaVersion: Schema.Literal(COMPACT_THREAD_CACHE_SCHEMA_VERSION), snapshot: fields.snapshot.mapFields((snapshotFields) => ({ ...snapshotFields, - turnItemsOmitLocalVisible: Schema.optionalKey(Schema.Literal(true)), + // Required metadata prevents partial compact records from masquerading as full ones. + turnItemsOmitLocalVisible: Schema.Boolean, + turnItemCount: Schema.Int, + visibleTurnItemCount: Schema.Int, })), })); +class InvalidThreadCache extends Data.TaggedError("InvalidThreadCache") {} + const encodeCompact = Schema.encodeEffect(Schema.fromJsonString(StoredCompactThreadSnapshot)); const decodeCompatible = Schema.decodeUnknownEffect( Schema.fromJsonString( @@ -38,21 +43,60 @@ export const encodeStoredThreadSnapshot = Effect.fn("ThreadCache.encode")(functi return yield* encodeCompact({ ...stored, schemaVersion: COMPACT_THREAD_CACHE_SCHEMA_VERSION, - snapshot: compact - ? { ...snapshot, projection: compact, turnItemsOmitLocalVisible: true } - : snapshot, + snapshot: { + ...snapshot, + projection: compact ?? snapshot.projection, + turnItemsOmitLocalVisible: compact !== null, + turnItemCount: snapshot.projection.turnItems.length, + visibleTurnItemCount: snapshot.projection.visibleTurnItems.length, + }, }); }); 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; - const { turnItemsOmitLocalVisible, ...snapshot } = stored.snapshot; - return { - ...stored, - snapshot: { - ...snapshot, - projection: boundedSnapshotProjection({ ...snapshot, turnItemsOmitLocalVisible }), - }, - }; + const { turnItemsOmitLocalVisible, turnItemCount, visibleTurnItemCount, ...snapshot } = + stored.snapshot; + const { projection } = snapshot; + const threadId = projection.thread.id; + let previousPosition = -1; + const visibleIds = new Set(); + const validRows = projection.visibleTurnItems.every((row) => { + const valid = + row.sourceItemId === row.item.id && + (row.sourceThreadId === row.item.threadId || + (row.item.type === "fork" && + row.item.source.type === "run" && + row.sourceThreadId === row.item.source.threadId)) && + (row.visibility !== "local" || row.sourceThreadId === threadId) && + (row.visibility !== "inherited" || row.sourceThreadId !== threadId) && + row.position > previousPosition && + !visibleIds.has(row.sourceItemId); + previousPosition = row.position; + visibleIds.add(row.sourceItemId); + return valid; + }); + const restored = turnItemsOmitLocalVisible + ? restoreLocalVisibleTurnItems(projection) + : projection; + const itemIds = new Set(); + const validItems = restored.turnItems.every((item) => { + const valid = item.threadId === threadId && !itemIds.has(item.id); + itemIds.add(item.id); + return valid; + }); + if ( + stored.threadId !== threadId || + restored.turnItems.length !== turnItemCount || + projection.visibleTurnItems.length !== visibleTurnItemCount || + !validRows || + !validItems || + projection.visibleTurnItems.some( + (row) => row.sourceThreadId === threadId && !itemIds.has(row.sourceItemId), + ) + ) { + return yield* new InvalidThreadCache(); + } + return { ...stored, snapshot: { ...snapshot, projection: restored } }; }); From ca6980cf2491bdb96c588b2169a197e9cf910981 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 04:12:31 +0000 Subject: [PATCH 4/5] test(mobile): cover cross-thread cache item ID collisions --- .../environment-cache-store.test.ts | 20 ++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/apps/mobile/src/connection/environment-cache-store.test.ts b/apps/mobile/src/connection/environment-cache-store.test.ts index 6857bc56ec88..1f11005e1680 100644 --- a/apps/mobile/src/connection/environment-cache-store.test.ts +++ b/apps/mobile/src/connection/environment-cache-store.test.ts @@ -374,6 +374,7 @@ describe("mobile SQLite environment cache store", () => { ); 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"), @@ -404,7 +405,12 @@ describe("mobile SQLite environment cache store", () => { turnItems: [local.item, synthetic, original.projection.turnItems[1]!], visibleTurnItems: [ { ...local, position: 0 }, - { ...original.projection.visibleTurnItems[0]!, position: 1 }, + { + ...inherited, + position: 1, + sourceItemId: local.sourceItemId, + item: { ...inherited.item, id: local.item.id }, + }, { ...local, position: 2, @@ -434,6 +440,18 @@ describe("mobile SQLite environment cache store", () => { "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"); From 8e4dcde14c02061f11d4de36237857e4f31d86e7 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 04:14:02 +0000 Subject: [PATCH 5/5] fix(mobile): validate cache rows by composite source identity --- packages/client-runtime/src/platform/threadCacheEncoding.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/client-runtime/src/platform/threadCacheEncoding.ts b/packages/client-runtime/src/platform/threadCacheEncoding.ts index ad5ac74cc412..68d09a9d2a15 100644 --- a/packages/client-runtime/src/platform/threadCacheEncoding.ts +++ b/packages/client-runtime/src/platform/threadCacheEncoding.ts @@ -63,6 +63,7 @@ export const decodeStoredThreadSnapshot = Effect.fn("ThreadCache.decode")(functi let previousPosition = -1; const visibleIds = new Set(); const validRows = projection.visibleTurnItems.every((row) => { + const identity = JSON.stringify([row.sourceThreadId, row.sourceItemId]); const valid = row.sourceItemId === row.item.id && (row.sourceThreadId === row.item.threadId || @@ -72,9 +73,9 @@ export const decodeStoredThreadSnapshot = Effect.fn("ThreadCache.decode")(functi (row.visibility !== "local" || row.sourceThreadId === threadId) && (row.visibility !== "inherited" || row.sourceThreadId !== threadId) && row.position > previousPosition && - !visibleIds.has(row.sourceItemId); + !visibleIds.has(identity); previousPosition = row.position; - visibleIds.add(row.sourceItemId); + visibleIds.add(identity); return valid; }); const restored = turnItemsOmitLocalVisible