diff --git a/apps/web/src/connection/storage.ts b/apps/web/src/connection/storage.ts index b25abe442ba11..bb27d7ad8e89a 100644 --- a/apps/web/src/connection/storage.ts +++ b/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(); diff --git a/packages/client-runtime/src/platform/threadCacheEncoding.test.ts b/packages/client-runtime/src/platform/threadCacheEncoding.test.ts index c62b8fc105152..0bd0874ffbc55 100644 --- a/packages/client-runtime/src/platform/threadCacheEncoding.test.ts +++ b/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: 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" }); + }), + ); +}); diff --git a/packages/client-runtime/src/platform/threadCacheEncoding.ts b/packages/client-runtime/src/platform/threadCacheEncoding.ts index 6466763a4de1b..8f087385d57e0 100644 --- a/packages/client-runtime/src/platform/threadCacheEncoding.ts +++ b/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()( {}, ) {} +/** 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(); 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) } }; });