From 437259bbf829754cbcf258376ee15e9fff5fa990 Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Thu, 5 Nov 2026 08:51:29 +0100 Subject: [PATCH 1/2] perf(web): a thread loaded from the cache holds each item once Web and desktop now write the compact record mobile already uses, so a saved thread takes about half the space. Decoding also shares equal item copies and message bodies again, including in records saved before this change and in records that could not be compacted, so a thread from the cache costs what a thread from the server costs. A pending question's card survives the round trip whether it sits inside the visible window, older than it, or older with newer items after it, and a record with no card is still fetched again. --- apps/web/src/connection/storage.test.ts | 92 +++++++++- apps/web/src/connection/storage.ts | 9 +- .../src/platform/threadCacheEncoding.test.ts | 168 ++++++++++++++++++ .../src/platform/threadCacheEncoding.ts | 49 ++++- .../src/state/threads-sync.test.ts | 125 ++++++++++++- 5 files changed, 429 insertions(+), 14 deletions(-) create mode 100644 packages/client-runtime/src/platform/threadCacheEncoding.test.ts diff --git a/apps/web/src/connection/storage.test.ts b/apps/web/src/connection/storage.test.ts index 7afe2b1e8573f..b08f2f8763ef2 100644 --- a/apps/web/src/connection/storage.test.ts +++ b/apps/web/src/connection/storage.test.ts @@ -2,7 +2,7 @@ import { ConnectionTransientError, PrimaryConnectionTarget, } from "@t3tools/client-runtime/connection"; -import { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import { EnvironmentId, ThreadId, TurnItemId } from "@t3tools/contracts"; import { ConnectionCatalogDocument, Persistence } from "@t3tools/client-runtime/platform"; import { describe, expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; @@ -12,6 +12,7 @@ import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import { afterEach, vi } from "vite-plus/test"; +import { makeThreadProjectionFixture } from "../test-fixtures"; import * as ConnectionStorage from "./storage"; const emptyCatalog = { @@ -197,6 +198,95 @@ describe("IndexedDB connection closed without a close event", () => { ); }); +describe("thread cache records in IndexedDB", () => { + const environmentId = EnvironmentId.make("environment-1"); + const output = "a command output that should reach the disk once"; + + const stubDatabase = (values: Map) => { + const settle = (request: T) => { + queueMicrotask(() => request.dispatchEvent(new Event("success"))); + return request; + }; + const database = Object.assign(new EventTarget(), { + close: vi.fn(), + transaction: () => { + const transaction = new EventTarget(); + return Object.assign(transaction, { + error: null, + objectStore: () => ({ + get: (key: IDBValidKey) => + settle(Object.assign(new EventTarget(), { result: values.get(key), error: null })), + put: (value: unknown, key: IDBValidKey) => { + values.set(key, value); + queueMicrotask(() => transaction.dispatchEvent(new Event("complete"))); + }, + }), + }); + }, + }) as unknown as IDBDatabase; + vi.stubGlobal("window", {}); + vi.stubGlobal("indexedDB", { + open: () => { + const request = Object.assign(new EventTarget(), { result: database, error: null }); + queueMicrotask(() => request.dispatchEvent(new Event("success"))); + return request; + }, + }); + }; + + it.effect("saves each item once and loads it as one object", () => + Effect.gen(function* () { + const values = new Map(); + stubDatabase(values); + const fixture = makeThreadProjectionFixture(); + const threadId = fixture.thread.id; + const at = fixture.updatedAt; + const item = { + id: TurnItemId.make("command"), + type: "command_execution" as const, + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed" as const, + title: null, + input: "pwd", + output, + exitCode: 0, + startedAt: at, + completedAt: at, + updatedAt: at, + }; + const projection = { + ...fixture, + turnItems: [item], + visibleTurnItems: [ + { + position: 0, + visibility: "local" as const, + sourceThreadId: threadId, + sourceItemId: item.id, + item, + }, + ], + }; + yield* Effect.gen(function* () { + const cache = yield* Persistence.EnvironmentCacheStore; + yield* cache.saveThread(environmentId, { snapshotSequence: 2, projection }); + const raw = [...values.values()][0] as string; + expect(raw.split(output).length - 1).toBe(1); + const loaded = Option.getOrThrow(yield* cache.loadThread(environmentId, threadId)); + expect(loaded.projection.turnItems).toEqual([item]); + expect(loaded.projection.visibleTurnItems[0]!.item).toBe(loaded.projection.turnItems[0]); + }).pipe(Effect.provide(ConnectionStorage.layer)); + }), + ); +}); + describe("browser GitHub routing permissions", () => { it.effect("revokes across runtimes before storage events and resists stale catalog writes", () => Effect.gen(function* () { diff --git a/apps/web/src/connection/storage.ts b/apps/web/src/connection/storage.ts index 7d9db4afdc494..09939d8546a8d 100644 --- a/apps/web/src/connection/storage.ts +++ b/apps/web/src/connection/storage.ts @@ -3,10 +3,10 @@ import { type ConnectionCatalogDocument as ConnectionCatalogDocumentType, EMPTY_CONNECTION_CATALOG_DOCUMENT, ORCHESTRATION_CACHE_SCHEMA_VERSION, - ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, StoredOrchestrationShellSnapshot, - StoredOrchestrationThreadSnapshot, decodeOrDiscardOrchestrationCache, + decodeStoredThreadSnapshot, + encodeStoredThreadSnapshot, putRemoteDpopTokenInCatalog, registerConnectionInCatalog, removeCatalogValue, @@ -50,8 +50,6 @@ const VCS_REFS_STORE_NAME = "vcs-refs"; 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, @@ -69,8 +67,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 encodeStoredThreadSnapshot = Schema.encodeEffect(StoredThreadSnapshotJson); const decodeStoredServerConfig = Schema.decodeUnknownEffect(StoredServerConfigJson); const encodeStoredServerConfig = Schema.encodeEffect(StoredServerConfigJson); const decodeStoredVcsRefs = Schema.decodeUnknownEffect(StoredVcsRefsJson); @@ -845,7 +841,6 @@ export const layer = Layer.effectContext( saveThread: Effect.fn("storage.saveThread")( function* (environmentId, snapshot) { const encoded = yield* encodeStoredThreadSnapshot({ - schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, environmentId, threadId: snapshot.projection.thread.id, snapshot, 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 0000000000000..9793bcf3456d9 --- /dev/null +++ b/packages/client-runtime/src/platform/threadCacheEncoding.test.ts @@ -0,0 +1,168 @@ +import { + EnvironmentId, + MessageId, + TurnItemId, + type OrchestrationV2ThreadDetailSnapshot, + type OrchestrationV2ThreadProjection, +} from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; + +import { v2Now, v2Projection, v2ThreadId } from "../state/orchestrationV2TestFixtures.ts"; +import { + ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, + StoredOrchestrationThreadSnapshot, +} from "./orchestrationCache.ts"; +import { + COMPACT_THREAD_CACHE_SCHEMA_VERSION, + decodeStoredThreadSnapshot, + encodeStoredThreadSnapshot, +} from "./threadCacheEncoding.ts"; + +const ENVIRONMENT_ID = EnvironmentId.make("environment-1"); +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 projection: 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 }; +const stored = { environmentId: ENVIRONMENT_ID, threadId: v2ThreadId, snapshot }; +const encodeFull = Schema.encodeEffect(Schema.fromJsonString(StoredOrchestrationThreadSnapshot)); +const occurrences = (text: string, part: string) => text.split(part).length - 1; + +describe("thread cache records", () => { + it.effect("writes each item once", () => + Effect.gen(function* () { + const raw = yield* encodeStoredThreadSnapshot(stored); + expect(JSON.parse(raw).schemaVersion).toBe(COMPACT_THREAD_CACHE_SCHEMA_VERSION); + expect(occurrences(raw, COMMAND_OUTPUT)).toBe(1); + }), + ); + + it.effect("decodes to the same projection, with one object per item", () => + Effect.gen(function* () { + const decoded = yield* decodeStoredThreadSnapshot(yield* encodeStoredThreadSnapshot(stored)); + const { projection: loaded } = decoded.snapshot; + expect(loaded).toEqual(projection); + expect(loaded.visibleTurnItems[0]!.item).toBe(loaded.turnItems[0]); + expect(loaded.visibleTurnItems[1]!.item).toBe(loaded.turnItems[1]); + }), + ); + + 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 = { + ...projection, + turnItems: [answer, hidden, command], + }; + const raw = yield* encodeStoredThreadSnapshot({ + ...stored, + 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* encodeFull({ + schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, + ...stored, + }); + expect(occurrences(raw, COMMAND_OUTPUT)).toBe(2); + const { projection: loaded } = (yield* decodeStoredThreadSnapshot(raw)).snapshot; + expect(loaded).toEqual(projection); + 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* encodeFull({ + schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, + ...stored, + snapshot: { + ...snapshot, + projection: { + ...projection, + 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 f993ff3ec2038..1ce8f1e50ccdc 100644 --- a/packages/client-runtime/src/platform/threadCacheEncoding.ts +++ b/packages/client-runtime/src/platform/threadCacheEncoding.ts @@ -1,13 +1,23 @@ +import type { OrchestrationV2ThreadProjection } 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 Schema from "effect/Schema"; -import { StoredOrchestrationThreadSnapshot } from "./orchestrationCache.ts"; +import { + ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, + 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; const StoredCompactThreadSnapshot = StoredOrchestrationThreadSnapshot.mapFields((fields) => ({ @@ -27,6 +37,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 encodeCompact = Schema.encodeEffect(Schema.fromJsonString(StoredCompactThreadSnapshot)); const decodeCompatible = Schema.decodeUnknownEffect( Schema.fromJsonString( @@ -57,7 +86,15 @@ export const encodeStoredThreadSnapshot = Effect.fn("ThreadCache.encode")(functi 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; @@ -80,9 +117,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); @@ -101,5 +140,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) } }; }); diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 20adc0e05fa0e..b99d04accaeea 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -45,9 +45,17 @@ import { StoredOrchestrationThreadSnapshot, decodeOrDiscardOrchestrationCache, } from "../platform/orchestrationCache.ts"; +import { + decodeStoredThreadSnapshot, + encodeStoredThreadSnapshot, +} from "../platform/threadCacheEncoding.ts"; import * as RpcSession from "../rpc/session.ts"; import { v2Now, v2Projection, v2ThreadId } from "./orchestrationV2TestFixtures.ts"; -import { onThreadRequestsResolved, type ThreadRequestsResolved } from "./threadRequests.ts"; +import { + derivePendingThreadRequests, + onThreadRequestsResolved, + type ThreadRequestsResolved, +} from "./threadRequests.ts"; import { EMPTY_THREAD_HISTORY_META } from "./threadHistoryMerge.ts"; import * as ThreadHistoryController from "./threadHistoryController.ts"; import { @@ -442,6 +450,121 @@ describe("EnvironmentThreads", () => { }), ); + describe("a pending question in a saved record", () => { + const newerItem = (id: string, ordinal: number): OrchestrationV2TurnItem => ({ + id: TurnItemId.make(id), + threadId: THREAD_ID, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed", + title: null, + startedAt: v2Now, + completedAt: v2Now, + updatedAt: v2Now, + type: "command_execution", + input: "pwd", + output: "done", + exitCode: 0, + }); + const rowOf = (item: OrchestrationV2TurnItem, position: number) => ({ + position, + visibility: "local" as const, + sourceThreadId: THREAD_ID, + sourceItemId: item.id, + item, + }); + const roundTrip = (projection: OrchestrationV2ThreadProjection) => + Effect.gen(function* () { + const raw = yield* encodeStoredThreadSnapshot({ + environmentId: TARGET.environmentId, + threadId: THREAD_ID, + snapshot: { snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, projection }, + }); + const stored = yield* decodeStoredThreadSnapshot(raw); + return { raw: JSON.parse(raw), snapshot: stored.snapshot }; + }); + const newer = newerItem("newer", 2); + const newest = newerItem("newest", 3); + + // The compact record leaves visible items out of `turnItems`, and the card check reads them. + it.effect.each([ + [ + "inside the visible window", + [QUESTION_CARD, newer], + [rowOf(QUESTION_CARD, 0), rowOf(newer, 1)], + true, + ], + ["older than the window", [newer, QUESTION_CARD], [rowOf(newer, 0)], true], + [ + "older than the window, with items after it", + [newer, QUESTION_CARD, newest], + [rowOf(newer, 0), rowOf(newest, 1)], + false, + ], + ] as const)( + "stays answerable without a refetch when its card is %s", + ([, turnItems, rows, compact]) => + Effect.gen(function* () { + const projection = { + ...PROJECTION_WITH_QUESTION, + turnItems: [...turnItems], + visibleTurnItems: [...rows], + }; + const stored = yield* roundTrip(projection); + expect(stored.raw.snapshot.turnItemsOmitLocalVisible).toBe(compact); + const h = yield* makeHarness({ + loadCached: Effect.succeedSome(stored.snapshot), + completionMarker: true, + }); + yield* Queue.offer(h.inputs, synchronized()); + const state = yield* awaitThreadState(h.observed, (value) => value.status === "live"); + expect(yield* Ref.get(h.loaderCalls)).toBe(0); + const loaded = Option.getOrThrow(state.data); + expect(loaded).toEqual(projection); + expect( + derivePendingThreadRequests(loaded).userInputs.map((input) => input.requestId), + ).toEqual([PENDING_QUESTION.id]); + for (const row of loaded.visibleTurnItems) { + expect(loaded.turnItems).toContain(row.item); + } + }), + ); + + it.effect("is fetched again when the record has no card for it", () => + Effect.gen(function* () { + const stored = yield* roundTrip({ + ...PROJECTION_WITH_QUESTION, + turnItems: [newer], + visibleTurnItems: [rowOf(newer, 0)], + }); + const h = yield* makeHarness({ + loadCached: Effect.succeedSome(stored.snapshot), + httpSnapshot: { + _tag: "present", + snapshot: { + snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, + projection: PROJECTION_WITH_QUESTION, + }, + }, + completionMarker: true, + }); + yield* Queue.offer(h.inputs, synchronized()); + const state = yield* awaitThreadState(h.observed, (value) => value.status === "live"); + expect(yield* Ref.get(h.loaderCalls)).toBe(1); + expect( + derivePendingThreadRequests(Option.getOrThrow(state.data)).userInputs.map( + (input) => input.requestId, + ), + ).toEqual([PENDING_QUESTION.id]); + }), + ); + }); + it.effect("discards an oversized version-4 cache and requests a fresh bounded snapshot", () => Effect.gen(function* () { const item: OrchestrationV2TurnItem = { From a2388a41d6a96db7a2517dea24aa9e5e42e38fa3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 15 Nov 2026 02:12:37 +0000 Subject: [PATCH 2/2] test(client-runtime): keep only the thread cache tests that fail without decode sharing The storage and thread-sync cases repeated the compact round-trip one layer up; mobile already exercised that path. Co-Authored-By: Claude Opus 5.5 --- apps/web/src/connection/storage.test.ts | 92 +------------ .../src/platform/threadCacheEncoding.test.ts | 24 +--- .../src/state/threads-sync.test.ts | 127 +----------------- 3 files changed, 3 insertions(+), 240 deletions(-) diff --git a/apps/web/src/connection/storage.test.ts b/apps/web/src/connection/storage.test.ts index 86c8be4a1ddce..1778111d3f4cb 100644 --- a/apps/web/src/connection/storage.test.ts +++ b/apps/web/src/connection/storage.test.ts @@ -2,7 +2,7 @@ import { ConnectionTransientError, PrimaryConnectionTarget, } from "@t3tools/client-runtime/connection"; -import { EnvironmentId, ThreadId, TurnItemId } from "@t3tools/contracts"; +import { EnvironmentId, ThreadId } from "@t3tools/contracts"; import { ConnectionCatalogDocument, Persistence } from "@t3tools/client-runtime/platform"; import { describe, expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; @@ -12,7 +12,6 @@ import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import { afterEach, vi } from "vite-plus/test"; -import { makeThreadProjectionFixture } from "../test-fixtures"; import * as ConnectionStorage from "./storage"; const emptyCatalog = { @@ -198,95 +197,6 @@ describe("IndexedDB connection closed without a close event", () => { ); }); -describe("thread cache records in IndexedDB", () => { - const environmentId = EnvironmentId.make("environment-1"); - const output = "a command output that should reach the disk once"; - - const stubDatabase = (values: Map) => { - const settle = (request: T) => { - queueMicrotask(() => request.dispatchEvent(new Event("success"))); - return request; - }; - const database = Object.assign(new EventTarget(), { - close: vi.fn(), - transaction: () => { - const transaction = new EventTarget(); - return Object.assign(transaction, { - error: null, - objectStore: () => ({ - get: (key: IDBValidKey) => - settle(Object.assign(new EventTarget(), { result: values.get(key), error: null })), - put: (value: unknown, key: IDBValidKey) => { - values.set(key, value); - queueMicrotask(() => transaction.dispatchEvent(new Event("complete"))); - }, - }), - }); - }, - }) as unknown as IDBDatabase; - vi.stubGlobal("window", {}); - vi.stubGlobal("indexedDB", { - open: () => { - const request = Object.assign(new EventTarget(), { result: database, error: null }); - queueMicrotask(() => request.dispatchEvent(new Event("success"))); - return request; - }, - }); - }; - - it.effect("saves each item once and loads it as one object", () => - Effect.gen(function* () { - const values = new Map(); - stubDatabase(values); - const fixture = makeThreadProjectionFixture(); - const threadId = fixture.thread.id; - const at = fixture.updatedAt; - const item = { - id: TurnItemId.make("command"), - type: "command_execution" as const, - threadId, - runId: null, - nodeId: null, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: 1, - status: "completed" as const, - title: null, - input: "pwd", - output, - exitCode: 0, - startedAt: at, - completedAt: at, - updatedAt: at, - }; - const projection = { - ...fixture, - turnItems: [item], - visibleTurnItems: [ - { - position: 0, - visibility: "local" as const, - sourceThreadId: threadId, - sourceItemId: item.id, - item, - }, - ], - }; - yield* Effect.gen(function* () { - const cache = yield* Persistence.EnvironmentCacheStore; - yield* cache.saveThread(environmentId, { snapshotSequence: 2, projection }); - const raw = [...values.values()][0] as string; - expect(raw.split(output).length - 1).toBe(1); - const loaded = Option.getOrThrow(yield* cache.loadThread(environmentId, threadId)); - expect(loaded.projection.turnItems).toEqual([item]); - expect(loaded.projection.visibleTurnItems[0]!.item).toBe(loaded.projection.turnItems[0]); - }).pipe(Effect.provide(ConnectionStorage.layer)); - }), - ); -}); - describe("browser GitHub routing permissions", () => { it.effect("revokes across runtimes before storage events and resists stale catalog writes", () => Effect.gen(function* () { diff --git a/packages/client-runtime/src/platform/threadCacheEncoding.test.ts b/packages/client-runtime/src/platform/threadCacheEncoding.test.ts index 959b2417a499c..0bd0874ffbc55 100644 --- a/packages/client-runtime/src/platform/threadCacheEncoding.test.ts +++ b/packages/client-runtime/src/platform/threadCacheEncoding.test.ts @@ -18,7 +18,6 @@ import { StoredOrchestrationThreadSnapshot, } from "./orchestrationCache.ts"; import { - COMPACT_THREAD_CACHE_SCHEMA_VERSION, decodeStoredThreadSnapshot, makeStoredThreadSnapshotEncoder, } from "./threadCacheEncoding.ts"; @@ -144,7 +143,6 @@ describe("cooperative thread cache encoding", () => { }); describe("thread cache records", () => { - const ENVIRONMENT_ID = EnvironmentId.make("environment-1"); const BODY = "a body long enough to count ".repeat(4); const COMMAND_OUTPUT = "only the items carry this output"; @@ -212,32 +210,13 @@ describe("thread cache records", () => { snapshotSequence: 3, projection: bodyProjection, }; - const record = { environmentId: ENVIRONMENT_ID, threadId: v2ThreadId, snapshot }; - const occurrences = (text: string, part: string) => text.split(part).length - 1; + const record = { environmentId, threadId: v2ThreadId, snapshot }; const encodeStoredThreadSnapshot = (input: typeof record) => makeStoredThreadSnapshotEncoder({ format: "compact" })(input).pipe( Effect.map(Option.getOrThrow), ); - it.effect("writes each item once", () => - Effect.gen(function* () { - const raw = yield* encodeStoredThreadSnapshot(record); - expect(JSON.parse(raw).schemaVersion).toBe(COMPACT_THREAD_CACHE_SCHEMA_VERSION); - expect(occurrences(raw, COMMAND_OUTPUT)).toBe(1); - }), - ); - - it.effect("decodes to the same projection, with one object per item", () => - Effect.gen(function* () { - const decoded = yield* decodeStoredThreadSnapshot(yield* encodeStoredThreadSnapshot(record)); - const { projection: loaded } = decoded.snapshot; - expect(loaded).toEqual(bodyProjection); - expect(loaded.visibleTurnItems[0]!.item).toBe(loaded.turnItems[0]); - expect(loaded.visibleTurnItems[1]!.item).toBe(loaded.turnItems[1]); - }), - ); - 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. @@ -269,7 +248,6 @@ describe("thread cache records", () => { schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, ...record, }); - expect(occurrences(raw, COMMAND_OUTPUT)).toBe(2); const { projection: loaded } = (yield* decodeStoredThreadSnapshot(raw)).snapshot; expect(loaded).toEqual(bodyProjection); expect(loaded.visibleTurnItems[1]!.item).toBe(loaded.turnItems[1]); diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 750ddfbeee3d9..5a771646f03fe 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -47,17 +47,9 @@ import { StoredOrchestrationThreadSnapshot, decodeOrDiscardOrchestrationCache, } from "../platform/orchestrationCache.ts"; -import { - decodeStoredThreadSnapshot, - makeStoredThreadSnapshotEncoder, -} from "../platform/threadCacheEncoding.ts"; import * as RpcSession from "../rpc/session.ts"; import { v2Now, v2Projection, v2ThreadId } from "./orchestrationV2TestFixtures.ts"; -import { - derivePendingThreadRequests, - onThreadRequestsResolved, - type ThreadRequestsResolved, -} from "./threadRequests.ts"; +import { onThreadRequestsResolved, type ThreadRequestsResolved } from "./threadRequests.ts"; import { EMPTY_THREAD_HISTORY_META } from "./threadHistoryMerge.ts"; import * as ThreadHistoryController from "./threadHistoryController.ts"; import { @@ -472,123 +464,6 @@ describe("EnvironmentThreads", () => { }), ); - describe("a pending question in a saved record", () => { - const newerItem = (id: string, ordinal: number): OrchestrationV2TurnItem => ({ - id: TurnItemId.make(id), - threadId: THREAD_ID, - runId: null, - nodeId: null, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal, - status: "completed", - title: null, - startedAt: v2Now, - completedAt: v2Now, - updatedAt: v2Now, - type: "command_execution", - input: "pwd", - output: "done", - exitCode: 0, - }); - const rowOf = (item: OrchestrationV2TurnItem, position: number) => ({ - position, - visibility: "local" as const, - sourceThreadId: THREAD_ID, - sourceItemId: item.id, - item, - }); - const roundTrip = (projection: OrchestrationV2ThreadProjection) => - Effect.gen(function* () { - const raw = Option.getOrThrow( - yield* makeStoredThreadSnapshotEncoder({ format: "compact" })({ - environmentId: TARGET.environmentId, - threadId: THREAD_ID, - snapshot: { snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, projection }, - }), - ); - const stored = yield* decodeStoredThreadSnapshot(raw); - return { raw: JSON.parse(raw), snapshot: stored.snapshot }; - }); - const newer = newerItem("newer", 2); - const newest = newerItem("newest", 3); - - // The compact record leaves visible items out of `turnItems`, and the card check reads them. - it.effect.each([ - [ - "inside the visible window", - [QUESTION_CARD, newer], - [rowOf(QUESTION_CARD, 0), rowOf(newer, 1)], - true, - ], - ["older than the window", [newer, QUESTION_CARD], [rowOf(newer, 0)], true], - [ - "older than the window, with items after it", - [newer, QUESTION_CARD, newest], - [rowOf(newer, 0), rowOf(newest, 1)], - false, - ], - ] as const)( - "stays answerable without a refetch when its card is %s", - ([, turnItems, rows, compact]) => - Effect.gen(function* () { - const projection = { - ...PROJECTION_WITH_QUESTION, - turnItems: [...turnItems], - visibleTurnItems: [...rows], - }; - const stored = yield* roundTrip(projection); - expect(stored.raw.snapshot.turnItemsOmitLocalVisible).toBe(compact); - const h = yield* makeHarness({ - loadCached: Effect.succeedSome(stored.snapshot), - completionMarker: true, - }); - yield* Queue.offer(h.inputs, synchronized()); - const state = yield* awaitThreadState(h.observed, (value) => value.status === "live"); - expect(yield* Ref.get(h.loaderCalls)).toBe(0); - const loaded = Option.getOrThrow(state.data); - expect(loaded).toEqual(projection); - expect( - derivePendingThreadRequests(loaded).userInputs.map((input) => input.requestId), - ).toEqual([PENDING_QUESTION.id]); - for (const row of loaded.visibleTurnItems) { - expect(loaded.turnItems).toContain(row.item); - } - }), - ); - - it.effect("is fetched again when the record has no card for it", () => - Effect.gen(function* () { - const stored = yield* roundTrip({ - ...PROJECTION_WITH_QUESTION, - turnItems: [newer], - visibleTurnItems: [rowOf(newer, 0)], - }); - const h = yield* makeHarness({ - loadCached: Effect.succeedSome(stored.snapshot), - httpSnapshot: { - _tag: "present", - snapshot: { - snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, - projection: PROJECTION_WITH_QUESTION, - }, - }, - completionMarker: true, - }); - yield* Queue.offer(h.inputs, synchronized()); - const state = yield* awaitThreadState(h.observed, (value) => value.status === "live"); - expect(yield* Ref.get(h.loaderCalls)).toBe(1); - expect( - derivePendingThreadRequests(Option.getOrThrow(state.data)).userInputs.map( - (input) => input.requestId, - ), - ).toEqual([PENDING_QUESTION.id]); - }), - ); - }); - it.effect("discards an oversized version-4 cache and requests a fresh bounded snapshot", () => Effect.gen(function* () { const item: OrchestrationV2TurnItem = {