Skip to content

Commit b76a110

Browse files
juliusmarmingejuliusmarminge
authored
perf(mobile): compact persisted thread snapshots (pingdotgg#24265)
1 parent d5b921c commit b76a110

5 files changed

Lines changed: 426 additions & 36 deletions

File tree

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

Lines changed: 294 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,17 @@
11
import {
22
ORCHESTRATION_CACHE_SCHEMA_VERSION,
33
ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
4+
COMPACT_THREAD_CACHE_SCHEMA_VERSION,
5+
StoredOrchestrationThreadSnapshot,
46
} from "@t3tools/client-runtime/platform";
57
import {
68
CommandId,
79
EnvironmentId,
810
ProjectId,
911
ProviderInstanceId,
12+
RunId,
1013
ThreadId,
14+
TurnItemId,
1115
type OrchestrationV2ShellSnapshot,
1216
type OrchestrationV2ThreadDetailSnapshot,
1317
type ThreadPullRequestLink,
@@ -17,6 +21,7 @@ import * as DateTime from "effect/DateTime";
1721
import * as Effect from "effect/Effect";
1822
import * as Logger from "effect/Logger";
1923
import * as Option from "effect/Option";
24+
import * as Schema from "effect/Schema";
2025

2126
import * as MobileDatabase from "../persistence/mobile-database";
2227
import { make } from "./environment-cache-store";
@@ -26,6 +31,9 @@ const PROJECT_ID = ProjectId.make("project-1");
2631
const THREAD_ID = ThreadId.make("thread-1");
2732
const PROVIDER_INSTANCE_ID = ProviderInstanceId.make("codex");
2833
const NOW = DateTime.makeUnsafe("2026-07-29T12:00:00.000Z");
34+
const encodeLegacyThreadSnapshot = Schema.encodeEffect(
35+
Schema.fromJsonString(StoredOrchestrationThreadSnapshot),
36+
);
2937
const SHELL_SNAPSHOT: OrchestrationV2ShellSnapshot = {
3038
schemaVersion: 1,
3139
snapshotSequence: 3,
@@ -152,6 +160,59 @@ function cacheId(
152160
return `${environmentId}:${kind}:${cacheKey}`;
153161
}
154162

163+
function threadSnapshotWithItems(): OrchestrationV2ThreadDetailSnapshot {
164+
const local = {
165+
id: TurnItemId.make("local-notice"),
166+
threadId: THREAD_ID,
167+
runId: null,
168+
nodeId: null,
169+
providerThreadId: null,
170+
providerTurnId: null,
171+
nativeItemRef: null,
172+
parentItemId: null,
173+
ordinal: 10,
174+
status: "completed" as const,
175+
title: null,
176+
startedAt: NOW,
177+
completedAt: NOW,
178+
updatedAt: NOW,
179+
type: "system_notice" as const,
180+
message: "A settled local notice.",
181+
};
182+
const hidden = { ...local, id: TurnItemId.make("hidden-notice"), ordinal: 11 };
183+
const inherited = {
184+
...local,
185+
id: TurnItemId.make("inherited-notice"),
186+
threadId: ThreadId.make("parent-thread"),
187+
};
188+
return {
189+
...THREAD_SNAPSHOT,
190+
historyCursor: "older-page",
191+
hasMoreHistory: true,
192+
latestLocalTurnOrdinal: 11,
193+
projection: {
194+
...THREAD_SNAPSHOT.projection,
195+
turnItems: [local, hidden],
196+
visibleTurnItems: [
197+
{
198+
position: 0,
199+
visibility: "inherited",
200+
sourceThreadId: inherited.threadId,
201+
sourceItemId: inherited.id,
202+
item: inherited,
203+
},
204+
{
205+
position: 1,
206+
visibility: "local",
207+
sourceThreadId: THREAD_ID,
208+
sourceItemId: local.id,
209+
item: local,
210+
},
211+
],
212+
},
213+
};
214+
}
215+
155216
function makeDatabase() {
156217
const values = new Map<string, string>();
157218
const schemaVersions = new Map<string, number>();
@@ -198,6 +259,238 @@ function makeDatabase() {
198259
}
199260

200261
describe("mobile SQLite environment cache store", () => {
262+
it.effect("loads a V5 record and migrates to V6 on the next save without losing items", () =>
263+
Effect.gen(function* () {
264+
const memory = makeDatabase();
265+
const store = yield* make().pipe(
266+
Effect.provideService(MobileDatabase.MobileDatabase, memory.database),
267+
);
268+
const snapshot = threadSnapshotWithItems();
269+
const id = cacheId(ENVIRONMENT_ID, "thread", THREAD_ID);
270+
const raw = yield* encodeLegacyThreadSnapshot({
271+
schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
272+
environmentId: ENVIRONMENT_ID,
273+
threadId: THREAD_ID,
274+
snapshot,
275+
});
276+
memory.values.set(id, raw);
277+
const loaded = Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID));
278+
expect(loaded).toEqual(snapshot);
279+
expect(memory.values.get(id)).toBe(raw);
280+
yield* store.saveThread(ENVIRONMENT_ID, loaded);
281+
expect(memory.schemaVersions.get(id)).toBe(COMPACT_THREAD_CACHE_SCHEMA_VERSION);
282+
expect(Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID))).toEqual(
283+
snapshot,
284+
);
285+
expect(memory.removed).toEqual([]);
286+
}),
287+
);
288+
289+
it.effect("discards corrupt or partial thread records so they can be refetched", () =>
290+
Effect.gen(function* () {
291+
const memory = makeDatabase();
292+
const store = yield* make().pipe(
293+
Effect.provideService(MobileDatabase.MobileDatabase, memory.database),
294+
);
295+
const id = cacheId(ENVIRONMENT_ID, "thread", THREAD_ID);
296+
yield* store.saveThread(ENVIRONMENT_ID, threadSnapshotWithItems());
297+
const raw = memory.values.get(id)!;
298+
const partial = JSON.parse(raw);
299+
delete partial.snapshot.projection.visibleTurnItems;
300+
const invalidItem = JSON.parse(raw);
301+
delete invalidItem.snapshot.projection.visibleTurnItems[1].item.message;
302+
const corruptions = [
303+
partial,
304+
invalidItem,
305+
...[
306+
(value: typeof partial) => {
307+
delete value.snapshot.turnItemCount;
308+
},
309+
(value: typeof partial) => {
310+
delete value.snapshot.turnItemsOmitLocalVisible;
311+
},
312+
(value: typeof partial) => {
313+
value.snapshot.turnItemsOmitLocalVisible = false;
314+
},
315+
(value: typeof partial) => {
316+
value.snapshot.projection.visibleTurnItems.pop();
317+
},
318+
(value: typeof partial) => {
319+
value.snapshot.projection.visibleTurnItems.shift();
320+
},
321+
(value: typeof partial) => {
322+
value.snapshot.projection.turnItems.pop();
323+
},
324+
(value: typeof partial) => {
325+
value.snapshot.turnItemCount = 0;
326+
},
327+
(value: typeof partial) => {
328+
value.snapshot.projection.visibleTurnItems[1].sourceItemId = "wrong-item";
329+
},
330+
(value: typeof partial) => {
331+
value.snapshot.projection.visibleTurnItems[1].visibility = "inherited";
332+
},
333+
(value: typeof partial) => {
334+
value.snapshot.projection.visibleTurnItems[1].item.threadId = "wrong-thread";
335+
},
336+
(value: typeof partial) => {
337+
value.snapshot.projection.thread.id = "wrong-thread";
338+
},
339+
(value: typeof partial) => {
340+
value.snapshot.projection.turnItems.push(
341+
value.snapshot.projection.visibleTurnItems[1].item,
342+
);
343+
},
344+
(value: typeof partial) => {
345+
value.snapshot.projection.visibleTurnItems.push(
346+
value.snapshot.projection.visibleTurnItems[1],
347+
);
348+
},
349+
].map((corrupt) => {
350+
const value = JSON.parse(raw);
351+
corrupt(value);
352+
return value;
353+
}),
354+
];
355+
const invalidRecords = ["{broken", ...corruptions.map((value) => JSON.stringify(value))];
356+
for (const invalid of invalidRecords) {
357+
memory.values.set(id, invalid);
358+
expect(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)).toEqual(Option.none());
359+
expect(memory.values.has(id)).toBe(false);
360+
}
361+
expect(memory.removed).toEqual(invalidRecords.map(() => id));
362+
yield* store.saveThread(ENVIRONMENT_ID, threadSnapshotWithItems());
363+
expect(Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID))).toEqual(
364+
threadSnapshotWithItems(),
365+
);
366+
}),
367+
);
368+
369+
it.effect("round-trips V5 exactly and restores shared items before returning to a consumer", () =>
370+
Effect.gen(function* () {
371+
const memory = makeDatabase();
372+
const store = yield* make().pipe(
373+
Effect.provideService(MobileDatabase.MobileDatabase, memory.database),
374+
);
375+
const original = threadSnapshotWithItems();
376+
const local = original.projection.visibleTurnItems[1]!;
377+
const inherited = original.projection.visibleTurnItems[0]!;
378+
const synthetic = { ...local.item, id: TurnItemId.make("synthetic-notice"), ordinal: 12 };
379+
const fork = {
380+
id: TurnItemId.make("fork-marker"),
381+
threadId: THREAD_ID,
382+
runId: null,
383+
nodeId: null,
384+
providerTurnId: null,
385+
nativeItemRef: null,
386+
parentItemId: null,
387+
ordinal: 0,
388+
status: "completed" as const,
389+
title: null,
390+
startedAt: null,
391+
completedAt: NOW,
392+
updatedAt: NOW,
393+
type: "fork" as const,
394+
source: {
395+
type: "run" as const,
396+
threadId: ThreadId.make("parent-thread"),
397+
runId: RunId.make("parent-run"),
398+
},
399+
targetThreadId: THREAD_ID,
400+
};
401+
const snapshot = {
402+
...original,
403+
projection: {
404+
...original.projection,
405+
turnItems: [local.item, synthetic, original.projection.turnItems[1]!],
406+
visibleTurnItems: [
407+
{ ...local, position: 0 },
408+
{
409+
...inherited,
410+
position: 1,
411+
sourceItemId: local.sourceItemId,
412+
item: { ...inherited.item, id: local.item.id },
413+
},
414+
{
415+
...local,
416+
position: 2,
417+
visibility: "synthetic" as const,
418+
sourceItemId: synthetic.id,
419+
item: synthetic,
420+
},
421+
{
422+
position: 3,
423+
visibility: "synthetic" as const,
424+
sourceThreadId: fork.source.threadId,
425+
sourceItemId: fork.id,
426+
item: fork,
427+
},
428+
],
429+
},
430+
};
431+
yield* store.saveThread(ENVIRONMENT_ID, snapshot);
432+
const id = cacheId(ENVIRONMENT_ID, "thread", THREAD_ID);
433+
const encoded = JSON.parse(memory.values.get(id)!);
434+
expect(encoded.schemaVersion).toBe(6);
435+
expect(memory.schemaVersions.get(id)).toBe(6);
436+
expect(encoded.snapshot.turnItemsOmitLocalVisible).toBe(true);
437+
expect(encoded.snapshot.turnItemCount).toBe(3);
438+
expect(encoded.snapshot.visibleTurnItemCount).toBe(4);
439+
expect(encoded.snapshot.projection.turnItems.map((item: { id: string }) => item.id)).toEqual([
440+
"hidden-notice",
441+
]);
442+
443+
const duplicate = JSON.parse(memory.values.get(id)!);
444+
duplicate.snapshot.projection.visibleTurnItems.push({
445+
...duplicate.snapshot.projection.visibleTurnItems[1],
446+
position: 4,
447+
});
448+
duplicate.snapshot.visibleTurnItemCount += 1;
449+
memory.values.set(id, JSON.stringify(duplicate));
450+
expect(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)).toEqual(Option.none());
451+
expect(memory.values.has(id)).toBe(false);
452+
expect(memory.removed).toEqual([id]);
453+
yield* store.saveThread(ENVIRONMENT_ID, snapshot);
454+
455+
const loaded = Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID));
456+
expect(loaded).toEqual(snapshot);
457+
expect(loaded).not.toHaveProperty("turnItemsOmitLocalVisible");
458+
expect(loaded).not.toHaveProperty("turnItemCount");
459+
expect(loaded).not.toHaveProperty("visibleTurnItemCount");
460+
expect(loaded.projection.turnItems[0]).toBe(loaded.projection.visibleTurnItems[0]!.item);
461+
expect(loaded.projection.turnItems[1]).toBe(loaded.projection.visibleTurnItems[2]!.item);
462+
expect(snapshot.projection.turnItems).toHaveLength(3);
463+
}),
464+
);
465+
466+
it.effect("keeps a full projection when visible rows cannot reproduce the original items", () =>
467+
Effect.gen(function* () {
468+
const memory = makeDatabase();
469+
const store = yield* make().pipe(
470+
Effect.provideService(MobileDatabase.MobileDatabase, memory.database),
471+
);
472+
const original = threadSnapshotWithItems();
473+
const local = original.projection.turnItems[0]!;
474+
for (const turnItems of [
475+
[{ ...local }, original.projection.turnItems[1]!],
476+
[{ ...local, message: "Newer than the visible row." }, original.projection.turnItems[1]!],
477+
original.projection.turnItems.toReversed(),
478+
]) {
479+
const snapshot = { ...original, projection: { ...original.projection, turnItems } };
480+
yield* store.saveThread(ENVIRONMENT_ID, snapshot);
481+
const raw = memory.values.get(cacheId(ENVIRONMENT_ID, "thread", THREAD_ID))!;
482+
const encoded = JSON.parse(raw);
483+
expect(encoded.schemaVersion).toBe(6);
484+
expect(encoded.snapshot.turnItemsOmitLocalVisible).toBe(false);
485+
expect(encoded.snapshot.turnItemCount).toBe(2);
486+
expect(encoded.snapshot.projection.turnItems).toHaveLength(2);
487+
expect(Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID))).toEqual(
488+
snapshot,
489+
);
490+
}
491+
}),
492+
);
493+
201494
it.effect("round-trips V2 shell and thread DateTime fields with the shared cache schema", () =>
202495
Effect.gen(function* () {
203496
const memory = makeDatabase();
@@ -229,7 +522,7 @@ describe("mobile SQLite environment cache store", () => {
229522
ORCHESTRATION_CACHE_SCHEMA_VERSION,
230523
);
231524
expect(memory.schemaVersions.get(cacheId(ENVIRONMENT_ID, "thread", THREAD_ID))).toBe(
232-
ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
525+
COMPACT_THREAD_CACHE_SCHEMA_VERSION,
233526
);
234527
}),
235528
);

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

Lines changed: 4 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,8 @@
11
import {
22
ORCHESTRATION_CACHE_SCHEMA_VERSION,
3-
ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
4-
StoredOrchestrationThreadSnapshot,
3+
COMPACT_THREAD_CACHE_SCHEMA_VERSION,
4+
decodeStoredThreadSnapshot,
5+
encodeStoredThreadSnapshot,
56
Persistence,
67
makeStoredShellSnapshotEncoder,
78
} from "@t3tools/client-runtime/platform";
@@ -33,12 +34,6 @@ const StoredVcsRefs = Schema.Struct({
3334
refs: VcsListRefsResult,
3435
});
3536

36-
const decodeStoredThreadSnapshot = Schema.decodeUnknownEffect(
37-
Schema.fromJsonString(StoredOrchestrationThreadSnapshot),
38-
);
39-
const encodeStoredThreadSnapshot = Schema.encodeEffect(
40-
Schema.fromJsonString(StoredOrchestrationThreadSnapshot),
41-
);
4237
const decodeStoredServerConfig = Schema.decodeUnknownEffect(
4338
Schema.fromJsonString(StoredServerConfig),
4439
);
@@ -147,19 +142,12 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () {
147142
saveThread: Effect.fn("MobileEnvironmentCache.saveThread")(function* (environmentId, snapshot) {
148143
const threadId = snapshot.projection.thread.id;
149144
const payload = yield* encodeStoredThreadSnapshot({
150-
schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
151145
environmentId,
152146
threadId,
153147
snapshot,
154148
}).pipe(Effect.mapError((cause) => persistenceError("save-thread", cause)));
155149
yield* database
156-
.saveCache(
157-
environmentId,
158-
"thread",
159-
threadId,
160-
ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION,
161-
payload,
162-
)
150+
.saveCache(environmentId, "thread", threadId, COMPACT_THREAD_CACHE_SCHEMA_VERSION, payload)
163151
.pipe(Effect.mapError(mapDatabaseError("save-thread")));
164152
}),
165153
removeThread: Effect.fn("MobileEnvironmentCache.removeThread")((environmentId, threadId) =>

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,5 +4,6 @@ export * as Persistence from "./persistence.ts";
44
export { ConnectionPersistenceError } from "./persistence.ts";
55
export * from "./orchestrationCache.ts";
66
export * from "./shellCacheEncoding.ts";
7+
export * from "./threadCacheEncoding.ts";
78
export * as PlatformConnectionSource from "./source.ts";
89
export * from "./storageDocument.ts";

0 commit comments

Comments
 (0)