From 25918c937ba13dae196f4bb82c67c0f834923417 Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Wed, 4 Nov 2026 19:17:08 +0100 Subject: [PATCH 01/10] perf(web): ChatView no longer keeps every thread copy it has shown ChatView's composer focus callbacks were made in its render. The React Compiler memoizes them, and a closure keeps its render's whole scope, the thread projection included. That scope held the memoized callbacks of an earlier render, which kept that render's scope, and so on. So while ChatView stayed mounted, every copy of a thread it had received (each reopen that loads the thread again delivers a new one) stayed in memory. The callbacks now come from a small hook whose scope holds only the composer ref, which breaks the chain. Browser harness (whole app, fake environment, 20 threads x 600 items, forced GC), 10 switches between two threads that each deliver a new copy, median of 5 alternating runs per arm: heap growth 60.1 -> 6.3 MB; heap after the switches 432.5 -> 378.5 MB. --- apps/web/src/components/ChatView.tsx | 19 +++++------------ .../src/components/chat/useComposerFocus.ts | 21 +++++++++++++++++++ 2 files changed, 26 insertions(+), 14 deletions(-) create mode 100644 apps/web/src/components/chat/useComposerFocus.ts diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index d1225065aeeca..14d43e3cea58e 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -183,7 +183,6 @@ import { isEditableFocused } from "../lib/editableFocus"; import { DEFAULT_RESOLVED_KEYBINDINGS } from "@t3tools/shared/keybindings"; import { shortcutLabelForCommand } from "../keybindings"; import { cn } from "~/lib/utils"; -import { isModelPickerOpen } from "../modelPickerVisibility"; import { COLLAPSED_SIDEBAR_TITLEBAR_INSET_CLASS } from "~/workspaceTitlebar"; import { toastManager } from "./ui/toast"; import { projectScriptIdFromCommand } from "~/projectScripts"; @@ -345,6 +344,7 @@ import { } from "./ChatView.logic"; import { useLocalStorage } from "~/hooks/useLocalStorage"; import { useComposerHandleContext } from "../composerHandleContext"; +import { useComposerFocus } from "./chat/useComposerFocus"; import { clampFileAttachmentUploadBytes } from "@t3tools/client-runtime/state/attachments"; import { appAtomRegistry } from "../rpc/atomRegistry"; import { assetEnvironment } from "../state/assets"; @@ -1801,10 +1801,8 @@ export default function ChatView(props: ChatViewProps) { sendInFlightRef, }); - const focusComposer = () => { - composerRef.current?.focusAtEnd(); - }; - const focusComposerStable = useStableCallback(focusComposer); + const { focusComposer: focusComposerStable, scheduleComposerFocus: scheduleComposerFocusStable } = + useComposerFocus(composerRef); const canInterruptRunningThread = canOperateThread && deriveCanInterruptRunningThread(activeThread !== undefined, activeRuntime); const onInterrupt = async () => { @@ -1831,13 +1829,6 @@ export default function ChatView(props: ChatViewProps) { () => (embedded ? undefined : subscribeSnapShotComposerFocus(focusComposerStable)), [embedded, focusComposerStable], ); - const scheduleComposerFocus = () => { - window.requestAnimationFrame(() => { - // The model picker stays open through its own changes, so it keeps focus. - if (!isModelPickerOpen()) focusComposer(); - }); - }; - const scheduleComposerFocusStable = useStableCallback(scheduleComposerFocus); const insertArtifactTemplate = (template: CodexArtifactTemplate) => { const composer = composerRef.current; if (!composer) return; @@ -1852,7 +1843,7 @@ export default function ChatView(props: ChatViewProps) { }); return; } - scheduleComposerFocus(); + scheduleComposerFocusStable(); }; const insertArtifactTemplateStable = useStableCallback(insertArtifactTemplate); const sendAppMessage = async (text: string) => { @@ -2846,7 +2837,7 @@ export default function ChatView(props: ChatViewProps) { canOverrideServerThreadEnvMode, setPendingServerThreadEnvModeEntry, setPendingServerThreadStartFromOriginByThreadId, - scheduleComposerFocus, + scheduleComposerFocus: scheduleComposerFocusStable, }); const { onResume, isResuming } = useComposerResume({ environmentId, diff --git a/apps/web/src/components/chat/useComposerFocus.ts b/apps/web/src/components/chat/useComposerFocus.ts new file mode 100644 index 0000000000000..38158e26033a3 --- /dev/null +++ b/apps/web/src/components/chat/useComposerFocus.ts @@ -0,0 +1,21 @@ +import type { ComposerHandleRef } from "../../composerHandleContext"; +import { useStableCallback } from "../../hooks/useStableCallback"; +import { isModelPickerOpen } from "../../modelPickerVisibility"; + +/** + * Made outside ChatView's render: a memoized closure there keeps that render's scope, thread + * projection included, and the scope keeps the previous render's, so every thread copy + * ChatView showed stayed in memory until it unmounted. + */ +export function useComposerFocus(composerRef: ComposerHandleRef) { + const focusComposer = useStableCallback(() => { + composerRef.current?.focusAtEnd(); + }); + const scheduleComposerFocus = useStableCallback(() => { + window.requestAnimationFrame(() => { + // The model picker stays open through its own changes, so it keeps focus. + if (!isModelPickerOpen()) focusComposer(); + }); + }); + return { focusComposer, scheduleComposerFocus }; +} From 5fac75c784dc5324dcfe8f5995626df1b375dca7 Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Wed, 4 Nov 2026 23:19:51 +0100 Subject: [PATCH 02/10] test(web): a mounted ChatView lets go of the thread structures it replaced The first test holds a WeakRef to each thread structure ChatView has shown and forces a GC: on main the four oldest of six are still alive, kept by the chain of render scopes behind the composer focus callbacks. The second file pins useComposerFocus: callers keep one identity, focus reaches the composer's current handle, and an open model picker keeps focus. --- .../src/components/ChatView.browser.test.tsx | 65 +++++++++++++++++-- .../chat/useComposerFocus.browser.test.tsx | 65 +++++++++++++++++++ 2 files changed, 126 insertions(+), 4 deletions(-) create mode 100644 apps/web/src/components/chat/useComposerFocus.browser.test.tsx diff --git a/apps/web/src/components/ChatView.browser.test.tsx b/apps/web/src/components/ChatView.browser.test.tsx index 96eb837db4d69..6b3668099b02b 100644 --- a/apps/web/src/components/ChatView.browser.test.tsx +++ b/apps/web/src/components/ChatView.browser.test.tsx @@ -1,10 +1,10 @@ import { createMemoryHistory } from "@tanstack/react-router"; -import { scopeProjectRef } from "@t3tools/client-runtime/environment"; -import { ThreadId } from "@t3tools/contracts"; +import { scopeProjectRef, scopeThreadRef } from "@t3tools/client-runtime/environment"; +import { RunId, ThreadId } from "@t3tools/contracts"; import { DEFAULT_CLIENT_SETTINGS } from "@t3tools/contracts/settings"; import type * as Context from "effect/Context"; import { expect, it, vi } from "vite-plus/test"; -import { page } from "vite-plus/test/browser"; +import { cdp, page } from "vite-plus/test/browser"; import { AppRoot } from "~/AppRoot"; // The route loads ChatView lazily, and its first transform (React Compiler over ~4k lines) @@ -13,13 +13,18 @@ import "~/components/ChatView"; import { DraftId, useComposerDraftStore } from "~/composerDraftStore"; import { __setClientSettingsForTests } from "~/hooks/useSettings"; import { getRouter } from "~/router"; +import { appAtomRegistry } from "~/rpc/atomRegistry"; +import { environmentThreadDetails } from "~/state/threads"; import { ENVIRONMENT_ID, PROJECT_ID, + assistantItem, + emitThreadEvent, installFakeEnvironment, + threadWithHistory, type FakeDriver, } from "../../test/fakeEnvironment"; -import { render } from "../../test/render"; +import { act, render } from "../../test/render"; // The socket is the only fake layer; see test/appScenarios.renders.tsx. vi.mock("../../../../packages/client-runtime/src/connection/driver.ts", async (importOriginal) => { @@ -65,3 +70,55 @@ it("a new draft does not load its reserved thread before the server lists it", a // A 404 here marks the thread deleted, and the view never loads it once the send creates it. expect(threadRequests).toEqual([]); }); + +it("a mounted thread view lets go of the thread structures it has replaced", async () => { + // Items are shared with the fixture, so a reopen's snapshot also carries the earlier updates. + const history = threadWithHistory("updated", 3); + const items = [...history.items]; + const updated = { ...history, items }; + const other = threadWithHistory("other", 3); + using _environment = installFakeEnvironment([updated, other]); + __setClientSettingsForTests({ + ...DEFAULT_CLIENT_SETTINGS, + onboardingCompletedAt: "2026-01-01T00:00:00.000Z", + }); + const router = getRouter(createMemoryHistory({ initialEntries: ["/"] })); + await router.load(); + await using _app = await render(, { router: false }); + const open = async (threadId: ThreadId, text: string) => { + await act(() => + router.navigate({ + to: "/$environmentId/$threadId", + params: { environmentId: ENVIRONMENT_ID, threadId }, + }), + ); + await expect.element(page.getByText(text).first()).toBeVisible(); + }; + const structureAtom = environmentThreadDetails.quietThreadAtom( + scopeThreadRef(ENVIRONMENT_ID, updated.id), + ); + + // Each round, a new answer arrives while the thread is shown, then the user switches away. + const shown: Array> = []; + await open(other.id, "Answer 2 for other."); + for (let round = 0; round < 6; round++) { + await open(updated.id, round === 0 ? "Answer 2 for updated." : `Update ${round - 1}.`); + const update = assistantItem( + updated.id, + RunId.make("updated-run-2"), + `updated-update-${round}`, + 6 + round, + `Update ${round}.`, + ); + items.push(update); + await act(() => emitThreadEvent(updated.id, { type: "turn-item.updated", payload: update })); + await expect.element(page.getByText(`Update ${round}.`)).toBeVisible(); + shown.push(new WeakRef(appAtomRegistry.get(structureAtom)!.projection)); + await open(other.id, "Answer 2 for other."); + } + await cdp().send("HeapProfiler.collectGarbage"); + + // The thread keeps its latest structure, and the timeline may remember the one before it. + const kept = shown.slice(0, -2).flatMap((structure, round) => (structure.deref() ? [round] : [])); + expect(kept).toEqual([]); +}); diff --git a/apps/web/src/components/chat/useComposerFocus.browser.test.tsx b/apps/web/src/components/chat/useComposerFocus.browser.test.tsx new file mode 100644 index 0000000000000..e153c2669170e --- /dev/null +++ b/apps/web/src/components/chat/useComposerFocus.browser.test.tsx @@ -0,0 +1,65 @@ +import type { RefObject } from "react"; +import { expect, it, vi } from "vite-plus/test"; + +import type { ChatComposerHandle } from "./ChatComposer"; +import { registerModelPickerContent } from "../../modelPickerVisibility"; +import { render } from "../../../test/render"; +import { useComposerFocus } from "./useComposerFocus"; + +type Focus = ReturnType; + +function Probe(props: { + readonly composerRef: RefObject; + readonly onRender: (focus: Focus) => void; +}) { + props.onRender(useComposerFocus(props.composerRef)); + return null; +} + +const handle = (focusAtEnd: () => void) => ({ focusAtEnd }) as unknown as ChatComposerHandle; + +it("focuses the composer's current handle without changing identity", async () => { + const composerRef: RefObject = { current: null }; + const renders: Focus[] = []; + const onRender = (focus: Focus) => renders.push(focus); + await using screen = await render(); + + const replaced = vi.fn(); + const current = vi.fn(); + composerRef.current = handle(replaced); + // The composer replaces its handle as the thread changes, with no render here in between. + composerRef.current = handle(current); + renders.at(-1)!.focusComposer(); + expect(current).toHaveBeenCalledTimes(1); + + await screen.rerender(); + expect(renders.at(-1)!.focusComposer).toBe(renders[0]!.focusComposer); + expect(renders.at(-1)!.scheduleComposerFocus).toBe(renders[0]!.scheduleComposerFocus); + expect(replaced).not.toHaveBeenCalled(); +}); + +it("schedules focus for the next frame unless the model picker is open by then", async () => { + const focusAtEnd = vi.fn(); + const composerRef: RefObject = { current: handle(focusAtEnd) }; + const renders: Focus[] = []; + await using _screen = await render( + renders.push(focus)} />, + ); + const { scheduleComposerFocus } = renders.at(-1)!; + const frames: FrameRequestCallback[] = []; + using _frames = vi.spyOn(window, "requestAnimationFrame").mockImplementation((frame) => { + frames.push(frame); + return frames.length; + }); + + scheduleComposerFocus(); + expect(focusAtEnd).not.toHaveBeenCalled(); + frames.shift()!(0); + expect(focusAtEnd).toHaveBeenCalledTimes(1); + + scheduleComposerFocus(); + const closeModelPicker = registerModelPickerContent(); + frames.shift()!(0); + closeModelPicker(); + expect(focusAtEnd).toHaveBeenCalledTimes(1); +}); 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 03/10] 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 06c22d715124969f9288b71cb2978b7a32d9919b Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Thu, 5 Nov 2026 08:55:44 +0100 Subject: [PATCH 04/10] perf(web): the timeline stops pinning an old thread in two long-lived closures The scroll pill's debouncer and the thread find scroll listener both closed over a render scope for as long as the visit or the container lived, which kept that render's whole timeline in memory, whichever thread it showed. --- .../chat/useThreadFindNavigation.ts | 46 ++++++++++++------- .../src/components/chat/useTimelineFollow.ts | 19 +++++--- 2 files changed, 41 insertions(+), 24 deletions(-) diff --git a/apps/web/src/components/chat/useThreadFindNavigation.ts b/apps/web/src/components/chat/useThreadFindNavigation.ts index 054faa23201e0..a00ab50f2f87e 100644 --- a/apps/web/src/components/chat/useThreadFindNavigation.ts +++ b/apps/web/src/components/chat/useThreadFindNavigation.ts @@ -28,6 +28,34 @@ interface RevealState { positionedEntries: FindEntries | null; } +/** + * Out of the hook, so the listener it keeps for as long as the container lives cannot pin a render + * of the timeline, and the entries of whichever thread that render showed. + */ +function trackPagingPosition( + container: HTMLElement, + stateRef: RefObject, + readInputs: () => RevealInputs, +): () => void { + const onScroll = () => { + const state = stateRef.current; + const position = state.pagingPosition; + const range = state.activeRange; + if ( + !position || + position.entries !== readInputs().entries || + !range?.startContainer.isConnected + ) + return; + const rect = range.getBoundingClientRect(); + const viewport = container.getBoundingClientRect(); + if (rect.bottom < viewport.top || rect.top > viewport.bottom) state.pagingPosition = null; + else position.top = rect.top; + }; + container.addEventListener("scroll", onScroll, { passive: true }); + return () => container.removeEventListener("scroll", onScroll); +} + /** * Brings the active match into view. Async continuations re-enter with the * inputs current by then, and stop once the match changes or the list unmounts. @@ -229,23 +257,7 @@ export function useThreadFindNavigation({ // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- attaches a scroll listener to the timeline container to track the active match position useEffect(() => { if (!container) return; - const onScroll = () => { - const state = stateRef.current; - const position = state.pagingPosition; - const range = state.activeRange; - if ( - !position || - position.entries !== readInputs().entries || - !range?.startContainer.isConnected - ) - return; - const rect = range.getBoundingClientRect(); - const viewport = container.getBoundingClientRect(); - if (rect.bottom < viewport.top || rect.top > viewport.bottom) state.pagingPosition = null; - else position.top = rect.top; - }; - container.addEventListener("scroll", onScroll, { passive: true }); - return () => container.removeEventListener("scroll", onScroll); + return trackPagingPosition(container, stateRef, readInputs); }, [container]); const requestedPagesRef = useRef<{ key: string | null; pages: Set }>({ key: null, diff --git a/apps/web/src/components/chat/useTimelineFollow.ts b/apps/web/src/components/chat/useTimelineFollow.ts index e872b13404e5f..edd5aa19c09e1 100644 --- a/apps/web/src/components/chat/useTimelineFollow.ts +++ b/apps/web/src/components/chat/useTimelineFollow.ts @@ -58,16 +58,25 @@ interface TimelineFollowVisit { * current visit, so a thread switch leaves the old one behind instead of * resetting it. */ +// Out here so the debouncer it keeps cannot pin a render of the hook, and the timeline of whichever +// thread that render showed. function createVisitScrollState( visit: TimelineFollowVisit, - showScrollToBottom: (visit: TimelineFollowVisit) => void, + stateRef: { readonly current: { readonly visit: TimelineFollowVisit } | null }, + setPill: (pill: { readonly visit: TimelineFollowVisit; readonly show: boolean }) => void, ) { return { visit, // Debounce *showing* the scroll-to-bottom pill so it doesn't flash during // thread switches. LegendList fires scroll events with isAtEnd=false while // initialScrollAtEnd is settling; hiding is always immediate. - showScrollDebouncer: new Debouncer(() => showScrollToBottom(visit), { wait: 150 }), + showScrollDebouncer: new Debouncer( + () => { + // A debounce that outlived its visit must not overwrite the current one's pill. + if (stateRef.current?.visit === visit) setPill({ visit, show: true }); + }, + { wait: 150 }, + ), positionedAnchor: null as MessageId | null, settledAnchor: null as MessageId | null, releaseAnchorScrollHold: null as (() => void) | null, @@ -223,11 +232,7 @@ export function useTimelineFollow(input: TimelineFollowInput) { /** Only from handlers: the bookkeeping of the visit they run in. */ const visitScroll = () => { if (visitScrollRef.current?.visit !== visit) { - visitScrollRef.current = createVisitScrollState(visit, (shownVisit) => { - // A debounce that outlived its visit must not overwrite the current one's pill. - if (visitScrollRef.current?.visit === shownVisit) - setPill({ visit: shownVisit, show: true }); - }); + visitScrollRef.current = createVisitScrollState(visit, visitScrollRef, setPill); } return visitScrollRef.current; }; From 969631cbdea3ce2a0491b4fee9cb3979aead66db Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Thu, 5 Nov 2026 14:13:20 +0100 Subject: [PATCH 05/10] perf(web): release left threads past a byte budget Threads the user navigated to and left may keep their state up to a byte budget, the largest released first. A released thread, by the budget or the idle TTL, keeps the end of its timeline for an immediate paint. Other mounts, unfinished runs, pending questions and approvals, unsaved threads and open threads stay outside the budget, and mobile sets no budget. A saved record that #26435 discards for a missing question card does not count as a copy to come back from. A thread the idle TTL released reopens from its record with the question still answerable. --- .../web/src/components/ChatView.logic.test.ts | 44 ++ apps/web/src/components/ChatView.logic.ts | 32 +- .../src/components/chat/ThreadTimeline.tsx | 10 +- apps/web/src/state/threads.ts | 20 +- .../src/state/threadRetention.test.ts | 48 ++ .../src/state/threadRetention.ts | 41 ++ .../src/state/threads-atoms.test.ts | 605 +++++++++++++++++- .../src/state/threads-sync.test.ts | 20 + packages/client-runtime/src/state/threads.ts | 130 +++- 9 files changed, 930 insertions(+), 20 deletions(-) create mode 100644 packages/client-runtime/src/state/threadRetention.test.ts diff --git a/apps/web/src/components/ChatView.logic.test.ts b/apps/web/src/components/ChatView.logic.test.ts index 81b6992a59d3b..efe68710b143d 100644 --- a/apps/web/src/components/ChatView.logic.test.ts +++ b/apps/web/src/components/ChatView.logic.test.ts @@ -32,6 +32,7 @@ import * as DateTime from "effect/DateTime"; import * as Option from "effect/Option"; import { afterEach, describe, expect, it, vi } from "vite-plus/test"; import { Atom } from "effect/reactivity"; +import { rememberTimelinePosition } from "./chat/timelineScrollAnchoring"; import { appAtomRegistry } from "../rpc/atomRegistry"; import { environmentThreadDetails } from "../state/threads"; @@ -42,6 +43,7 @@ import { resolveComposerProviderSelection, resolveWorktreeSetupProgress, peekRememberedThreadTimeline, + releaseRememberedThreadTimeline, rememberReadyThreadTimeline, resetHeldThreadTimeline, resolveThreadSwitchTimeline, @@ -423,6 +425,48 @@ describe("resolveThreadSwitchTimeline", () => { ).toEqual(["a1", "a2"]); }); + describe("a released thread", () => { + const entries = Array.from({ length: 200 }, (_, index) => `entry-${index}`); + const threadKey = "env-1:released"; + + it("keeps only the end of its remembered timeline", () => { + rememberReadyThreadTimeline({ threadKey, entries }); + releaseRememberedThreadTimeline(threadKey); + const kept = peekRememberedThreadTimeline(threadKey)!; + expect(kept.length).toBeLessThan(100); + expect(kept).toEqual(entries.slice(-kept.length)); + expect( + resolveThreadSwitchTimeline({ loading: true, activeThreadKey: threadKey, nextEntries: [] }), + ).toEqual(kept); + }); + + it("leaves other threads' remembered timelines whole", () => { + rememberReadyThreadTimeline({ threadKey: "env-1:other", entries }); + rememberReadyThreadTimeline({ threadKey, entries }); + releaseRememberedThreadTimeline(threadKey); + expect(peekRememberedThreadTimeline("env-1:other")).toEqual(entries); + }); + + it("keeps none when the reader's place is further up the timeline", () => { + rememberReadyThreadTimeline({ threadKey, entries }); + rememberTimelinePosition( + threadKey, + { rowId: "entry-3", offsetWithinRow: 0, scrollOffset: 0, atEnd: false }, + false, + ); + releaseRememberedThreadTimeline(threadKey); + expect(peekRememberedThreadTimeline(threadKey)).toBeNull(); + }); + + it("still remembers sixteen threads", () => { + const keys = Array.from({ length: 18 }, (_, index) => `env-1:thread-${index}`); + for (const key of keys) rememberReadyThreadTimeline({ threadKey: key, entries: [key] }); + expect(keys.map((key) => peekRememberedThreadTimeline(key) !== null)).toEqual( + keys.map((_, index) => index >= 2), + ); + }); + }); + it("does not paint the last-viewed thread for a destination with nothing remembered", () => { rememberReadyThreadTimeline(held); expect( diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index e933abf86b2fb..2d4efe2b39826 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -76,7 +76,12 @@ import { } from "../composerDraftStore"; import * as Schema from "effect/Schema"; import { appAtomRegistry } from "../rpc/atomRegistry"; -import { environmentThreadShells, environmentThreadDetails } from "../state/threads"; +import { + environmentThreadDetails, + environmentThreadShells, + onLeftThreadReleased, +} from "../state/threads"; +import { readTimelinePosition } from "./chat/timelineScrollAnchoring"; import { waitForAtomValue } from "../state/waitForAtomValue"; import { filterTerminalContextsWithText, @@ -383,6 +388,31 @@ export function peekRememberedThreadTimeline( return (rememberedThreadTimelines.get(threadKey)?.entries as T | undefined) ?? null; } +// Fills the pane at the end of a timeline until the thread is back from the cache. +const RELEASED_THREAD_TIMELINE_ENTRIES = 60; + +/** + * These rows point at the thread's items, so a released thread keeps only its last few. A reader's + * place further up can only be restored on the whole timeline, so that thread keeps none. + */ +export function releaseRememberedThreadTimeline(threadKey: string): void { + const held = rememberedThreadTimelines.get(threadKey); + if (held === undefined) return; + if (readTimelinePosition(threadKey)?.atEnd === false) { + rememberedThreadTimelines.delete(threadKey); + rememberedThreadTimelineOrder = rememberedThreadTimelineOrder.filter( + (key) => key !== threadKey, + ); + return; + } + rememberedThreadTimelines.set(threadKey, { + threadKey, + entries: held.entries.slice(-RELEASED_THREAD_TIMELINE_ENTRIES), + }); +} + +onLeftThreadReleased(releaseRememberedThreadTimeline); + export function resetHeldThreadTimeline(): void { rememberedThreadTimelines = new Map(); rememberedThreadTimelineOrder = []; diff --git a/apps/web/src/components/chat/ThreadTimeline.tsx b/apps/web/src/components/chat/ThreadTimeline.tsx index a68d494d8b0a0..02c43dfb486e6 100644 --- a/apps/web/src/components/chat/ThreadTimeline.tsx +++ b/apps/web/src/components/chat/ThreadTimeline.tsx @@ -1,14 +1,16 @@ import type { ScopedThreadRef } from "@t3tools/contracts"; -import { type ComponentProps, useLayoutEffect } from "react"; +import { type ComponentProps, use, useLayoutEffect } from "react"; import { useIncrementalTimelineEntries } from "../../hooks/useIncrementalTimelineEntries"; import type { TimelineEntriesInput, TimelineEntry } from "../../session-logic"; import { useTimelineVisibleTurnItems } from "../../state/entities"; +import { markThreadVisited } from "../../state/threads"; import { peekRememberedThreadTimeline, rememberReadyThreadTimeline, resolveThreadSwitchTimeline, timelineHasEphemeralPreviewUrls, } from "../ChatView.logic"; +import { BackgroundThreadContext } from "./backgroundThread"; import { MessagesTimeline } from "./MessagesTimeline"; /** @@ -47,6 +49,12 @@ export function ThreadTimeline({ nextEntries: entries, rememberedForActive: peekRememberedThreadTimeline(activeThreadKey), }); + // A side chat shows its thread beside the one the user navigated to. + const backgroundThread = use(BackgroundThreadContext); + // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- tells the left-thread budget which thread the user navigated to, once it is on screen + useLayoutEffect(() => { + if (!backgroundThread && threadDetailRef !== null) markThreadVisited(threadDetailRef); + }, [backgroundThread, threadDetailRef]); // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- stores the committed ready timeline in the module-level cache used during thread switches useLayoutEffect(() => { if (detailLoading || entries.length === 0 || timelineHasEphemeralPreviewUrls(entries)) { diff --git a/apps/web/src/state/threads.ts b/apps/web/src/state/threads.ts index 6d2ce2a07c90b..bc0e27d7622d7 100644 --- a/apps/web/src/state/threads.ts +++ b/apps/web/src/state/threads.ts @@ -1,6 +1,7 @@ import { createChildThreadApprovalsAtom } from "@t3tools/client-runtime/state/thread-requests"; import { useAtomValue } from "@effect/atom-react"; import { enabledEnvironmentIds } from "@t3tools/client-runtime/state/connections"; +import { scopedThreadKey } from "@t3tools/client-runtime/environment"; import { arrayElementsEqual } from "@t3tools/client-runtime/state/entities"; import { createEnvironmentThreadDetailAtoms, @@ -33,7 +34,24 @@ export const threadEnvironment = createThreadEnvironmentAtoms( connectionAtomRuntime, threadFamilies.snapshotAtom, ); -const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime); +// What left threads may keep in memory. A thread opened from a bounded server window costs a few MB +// at most, so only threads grown by long streaming or scrollback reach it. +const LEFT_THREADS_MAX_BYTES = 50 * 1024 * 1024; + +let leftThreadReleased: (threadKey: string) => void = () => {}; + +/** Whoever remembers more of a left thread lets go of it here, together with its state. */ +export function onLeftThreadReleased(listener: (threadKey: string) => void): void { + leftThreadReleased = listener; +} + +const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime, { + leftThreads: { + maxBytes: LEFT_THREADS_MAX_BYTES, + onRelease: (thread) => leftThreadReleased(scopedThreadKey(thread)), + }, +}); +export const markThreadVisited = environmentThreads.markThreadVisited; export const environmentThreadDetails = createEnvironmentThreadDetailAtoms( environmentThreads.stateAtom, ); diff --git a/packages/client-runtime/src/state/threadRetention.test.ts b/packages/client-runtime/src/state/threadRetention.test.ts new file mode 100644 index 0000000000000..6d6aae3df6f54 --- /dev/null +++ b/packages/client-runtime/src/state/threadRetention.test.ts @@ -0,0 +1,48 @@ +import type { OrchestrationV2ThreadProjection } from "@t3tools/contracts"; +import { describe, expect, it } from "vite-plus/test"; + +import { v2Projection } from "./orchestrationV2TestFixtures.ts"; +import { estimateRetainedThreadBytes } from "./threadRetention.ts"; + +const withItems = (count: number, bytes: number): OrchestrationV2ThreadProjection => { + const items = Array.from({ length: count }, (_, index) => ({ + id: `item-${index}`, + threadId: v2Projection.thread.id, + type: "system_notice", + message: "x".repeat(bytes), + })); + return { + ...v2Projection, + turnItems: items, + visibleTurnItems: items.map((item, position) => ({ + position, + visibility: "local", + sourceThreadId: v2Projection.thread.id, + sourceItemId: item.id, + item, + })), + } as unknown as OrchestrationV2ThreadProjection; +}; + +describe("estimateRetainedThreadBytes", () => { + it("counts text once for an item the timeline shares, and grows with the items", () => { + const small = estimateRetainedThreadBytes(withItems(10, 100)); + const large = estimateRetainedThreadBytes(withItems(10, 100_000)); + expect(large - small).toBe(10 * (100_000 - 100)); + expect(estimateRetainedThreadBytes(withItems(20, 100))).toBeGreaterThan(small * 1.9); + }); + + it("charges an inherited row, which no local item shares", () => { + const local = withItems(1, 1_000); + const inherited = { + ...local, + turnItems: [], + visibleTurnItems: local.visibleTurnItems.map((row) => ({ + ...row, + visibility: "inherited", + sourceThreadId: "parent", + })), + } as unknown as OrchestrationV2ThreadProjection; + expect(estimateRetainedThreadBytes(inherited)).toBeGreaterThan(1_000); + }); +}); diff --git a/packages/client-runtime/src/state/threadRetention.ts b/packages/client-runtime/src/state/threadRetention.ts index 3f2b8a6fd0166..9b6d0f3de490f 100644 --- a/packages/client-runtime/src/state/threadRetention.ts +++ b/packages/client-runtime/src/state/threadRetention.ts @@ -1 +1,42 @@ +import type { OrchestrationV2RunStatus, OrchestrationV2ThreadProjection } from "@t3tools/contracts"; + export const THREAD_SNAPSHOT_IDLE_TTL_MS = 5 * 60_000; + +/** + * Runs the user is likely to come back to. Preparing, starting and running ones also have newer + * state than the persisted cache, which is saved only once they settle. + */ +export const UNFINISHED_RUN_STATUSES: ReadonlySet = new Set([ + "preparing", + "queued", + "starting", + "running", + "waiting", +]); + +// What the heap charges a small object beyond its strings. +const OBJECT_OVERHEAD_BYTES = 64; + +function retainedBytes(value: unknown): number { + if (typeof value === "string") return value.length; + if (typeof value !== "object" || value === null) return 8; + let bytes = OBJECT_OVERHEAD_BYTES; + for (const field of Object.values(value)) bytes += retainedBytes(field); + return bytes; +} + +/** + * A cheap estimate of what a thread keeps in the heap, from its items, inherited rows and message + * records, with strings counted by length. It walks one thread when the thread is left. Timeline + * rows and parsed markdown come on top, which matters only for small threads. + */ +export function estimateRetainedThreadBytes(projection: OrchestrationV2ThreadProjection): number { + let bytes = retainedBytes(projection.turnItems) + retainedBytes(projection.messages); + for (const row of projection.visibleTurnItems) { + bytes += + row.sourceThreadId === projection.thread.id + ? OBJECT_OVERHEAD_BYTES + : retainedBytes(row.item) + OBJECT_OVERHEAD_BYTES; + } + return bytes; +} diff --git a/packages/client-runtime/src/state/threads-atoms.test.ts b/packages/client-runtime/src/state/threads-atoms.test.ts index af2ab6d0575a6..ab4501a32f78a 100644 --- a/packages/client-runtime/src/state/threads-atoms.test.ts +++ b/packages/client-runtime/src/state/threads-atoms.test.ts @@ -1,8 +1,15 @@ import { EnvironmentId, EventId, + MessageId, + NodeId, ORCHESTRATION_V2_WS_METHODS, + ProviderInstanceId, + RunId, + RuntimeRequestId, + ThreadId, TurnItemId, + type OrchestrationV2Run, type OrchestrationV2ThreadHistoryPage, type OrchestrationV2ThreadProjection, type OrchestrationV2ThreadDetailSnapshot, @@ -30,12 +37,22 @@ import { } from "../connection/model.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as Persistence from "../platform/persistence.ts"; +import { + decodeStoredThreadSnapshot, + encodeStoredThreadSnapshot, +} from "../platform/threadCacheEncoding.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; import type { RpcSession } from "../rpc/session.ts"; import { createEnvironmentThreadDetailAtoms } from "./threadDetail.ts"; import { v2Projection, v2ThreadId } from "./orchestrationV2TestFixtures.ts"; import * as ThreadHistoryController from "./threadHistoryController.ts"; -import { createEnvironmentThreadStateAtoms, type EnvironmentThreadState } from "./threads.ts"; +import { derivePendingThreadRequests } from "./threadRequests.ts"; +import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts"; +import { + createEnvironmentThreadStateAtoms, + type EnvironmentThreadState, + type LeftThreadRetention, +} from "./threads.ts"; import * as ThreadSnapshotLoader from "./threadSnapshotHttp.ts"; const TARGET = new PrimaryConnectionTarget({ @@ -54,8 +71,21 @@ const SNAPSHOT: OrchestrationV2ThreadDetailSnapshot = { snapshotSequence: 7, pro const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?: { readonly snapshot?: OrchestrationV2ThreadDetailSnapshot; readonly snapshotUnavailable?: boolean; + /** Serves each thread its own projection and keeps saved threads, like the real cache. */ + readonly projectionFor?: (threadId: ThreadId) => OrchestrationV2ThreadProjection; + readonly leftThreads?: LeftThreadRetention; + /** Saves as the real cache does, so a reopened thread has been through its record format. */ + readonly throughCacheRecord?: boolean; + /** Runs before each save, to hold one back or fail it. */ + readonly beforeSave?: ( + threadId: ThreadId, + ) => Effect.Effect; }) { + const savedThreads = new Map(); + // Offered once a thread state's scope has run all its finalizers. + const threadScopesClosed = yield* Queue.unbounded(); const subscriptions = yield* Queue.unbounded<{ + readonly threadId: ThreadId; readonly afterSequence: number | undefined; readonly events: Queue.Queue; readonly closed: Deferred.Deferred; @@ -71,7 +101,10 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? let opened = 0; let active = 0; const client = { - [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: (input: { readonly afterSequence?: number }) => + [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: (input: { + readonly threadId: ThreadId; + readonly afterSequence?: number; + }) => Stream.unwrap( Effect.gen(function* () { const events = yield* Queue.unbounded(); @@ -86,7 +119,12 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? active -= 1; }).pipe(Effect.andThen(Deferred.succeed(closed, undefined))), ); - yield* Queue.offer(subscriptions, { afterSequence: input.afterSequence, events, closed }); + yield* Queue.offer(subscriptions, { + threadId: input.threadId, + afterSequence: input.afterSequence, + events, + closed, + }); return Stream.fromQueue(events); }), ), @@ -121,6 +159,8 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? disconnect: Effect.void, retryNow: Effect.void, }); + // Each change replaces the environment's entry, like learning a route, which restarts its streams. + const environmentEntry = yield* SubscriptionRef.make(0); const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({ entries: yield* SubscriptionRef.make>( new Map(), @@ -143,8 +183,22 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor), runStream: (_environmentId, stream) => Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor), - followStream: (_environmentId, stream) => - Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + followStream: (_environmentId, stream) => { + const follow = Stream.provideService( + stream, + EnvironmentSupervisor.EnvironmentSupervisor, + supervisor, + ).pipe(Stream.ensuring(Queue.offer(threadScopesClosed, undefined))); + return options?.leftThreads === undefined + ? follow + : Stream.concat( + Stream.fromEffect(SubscriptionRef.get(environmentEntry)), + SubscriptionRef.changes(environmentEntry), + ).pipe( + Stream.changes, + Stream.switchMap(() => follow), + ); + }, }); const historyController = yield* Effect.service( ThreadHistoryController.ThreadHistoryController, @@ -169,13 +223,33 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? Persistence.EnvironmentCacheStore.of({ loadShell: () => Effect.succeedNone, saveShell: () => Effect.void, - loadThread: () => + loadThread: (_environmentId, threadId) => Effect.sync(() => { diskLoads += 1; - return Option.none(); + return Option.fromUndefinedOr(savedThreads.get(threadId)); + }), + saveThread: (_environmentId, saved) => + (options?.beforeSave?.(saved.projection.thread.id) ?? Effect.void).pipe( + Effect.andThen( + options?.throughCacheRecord === true + ? encodeStoredThreadSnapshot({ + environmentId: TARGET.environmentId, + threadId: saved.projection.thread.id, + snapshot: saved, + }).pipe(Effect.flatMap(decodeStoredThreadSnapshot), Effect.orDie) + : Effect.succeed({ snapshot: saved }), + ), + Effect.tap(({ snapshot: stored }) => + Effect.sync(() => { + if (options?.projectionFor) savedThreads.set(stored.projection.thread.id, stored); + }), + ), + Effect.asVoid, + ), + removeThread: (_environmentId, threadId) => + Effect.sync(() => { + savedThreads.delete(threadId); }), - saveThread: () => Effect.void, - removeThread: () => Effect.void, loadServerConfig: () => Effect.succeedNone, saveServerConfig: () => Effect.void, loadVcsRefs: () => Effect.succeedNone, @@ -188,7 +262,7 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? Layer.succeed( ThreadSnapshotLoader.ThreadSnapshotLoader, ThreadSnapshotLoader.ThreadSnapshotLoader.of({ - load: () => + load: (_prepared, threadId) => Effect.sync(() => { httpLoads += 1; if (options?.snapshotUnavailable === true) { @@ -196,7 +270,9 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? } return { _tag: "present" as const, - snapshot, + snapshot: options?.projectionFor + ? { ...snapshot, projection: options.projectionFor(threadId) } + : snapshot, ...(snapshot.historyCursor === undefined ? {} : { @@ -212,7 +288,10 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? ), ), ); - const raw = createEnvironmentThreadStateAtoms(runtime); + const raw = createEnvironmentThreadStateAtoms( + runtime, + options?.leftThreads && { leftThreads: options.leftThreads }, + ); const details = createEnvironmentThreadDetailAtoms(raw.stateAtom); const ref = { environmentId: TARGET.environmentId, threadId: THREAD_ID }; const stateAtom = details.stateAtom(ref); @@ -231,6 +310,8 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? ref, subscriptions, olderLoads, + threadScopesClosed, + restartEnvironment: SubscriptionRef.update(environmentEntry, (entry) => entry + 1), loadEarlier: () => historyController.loadEarlier(TARGET.environmentId, THREAD_ID), counts: () => ({ httpLoads, diskLoads, opened, active }), }; @@ -450,3 +531,503 @@ it.effect("keeps snapshot boundaries and completion markers ordered inside a rep yield* Deferred.await(next.closed); }), ); + +describe("left threads", () => { + type Harness = Effect.Success>; + const LEFT = Array.from({ length: 6 }, (_, index) => ThreadId.make(`left-${index}`)); + const BODY = 10_000; + // Three left threads of BODY fit, besides the one just left; four do not. + const MAX_BYTES = 35_000; + const at = THREAD.thread.createdAt; + const RUNNING: OrchestrationV2Run = { + id: RunId.make("left-run"), + threadId: LEFT[0]!, + ordinal: 1, + providerInstanceId: ProviderInstanceId.make("codex"), + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + providerThreadId: null, + userMessageId: MessageId.make("left-message"), + rootNodeId: null, + activeAttemptId: null, + status: "running", + requestedAt: at, + startedAt: at, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }; + const itemIdOf = (threadId: ThreadId) => TurnItemId.make(`${threadId}-notice`); + const itemsFor = (threadId: ThreadId, bytes: number) => { + const item = { + id: itemIdOf(threadId), + type: "system_notice" as const, + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed" as const, + title: null, + message: "x".repeat(bytes), + startedAt: at, + completedAt: at, + updatedAt: at, + }; + return { + turnItems: [item], + visibleTurnItems: [ + { + position: 0, + visibility: "local" as const, + sourceThreadId: threadId, + sourceItemId: item.id, + item, + }, + ], + }; + }; + const projectionFor = + (options?: { + readonly running?: ReadonlySet; + readonly bytes?: (threadId: ThreadId) => number; + }) => + (threadId: ThreadId): OrchestrationV2ThreadProjection => ({ + ...THREAD, + ...itemsFor(threadId, options?.bytes?.(threadId) ?? BODY), + thread: { ...THREAD.thread, id: threadId }, + runs: options?.running?.has(threadId) ? [{ ...RUNNING, threadId }] : [], + }); + const event = (threadId: ThreadId, deleted: boolean): OrchestrationV2ThreadStreamItem => { + const payload = { ...THREAD.thread, id: threadId, title: `Visited ${threadId}` }; + const common = { id: EventId.make(`left-${threadId}`), threadId, occurredAt: at }; + return { + kind: "event", + sequence: 8, + event: deleted + ? { ...common, type: "thread.deleted", payload: { ...payload, deletedAt: at } } + : { ...common, type: "thread.metadata-updated", payload }, + }; + }; + const refOf = (threadId: ThreadId) => ({ environmentId: TARGET.environmentId, threadId }); + const atomOf = (h: Harness, threadId: ThreadId) => h.details.stateAtom(refOf(threadId)); + const budget = (released: ThreadId[], maxBytes = MAX_BYTES): LeftThreadRetention => ({ + maxBytes, + onRelease: (thread) => released.push(thread.threadId), + }); + + /** Opens the thread, retitles it at sequence 8 (or deletes it), and returns how to leave it. */ + const open = Effect.fn("TestThreadAtoms.open")(function* ( + h: Harness, + threadId: ThreadId, + options?: { readonly deleted?: boolean; readonly secondary?: boolean }, + ) { + const deleted = options?.deleted === true; + const atom = atomOf(h, threadId); + if (options?.secondary !== true) h.rawAtoms.markThreadVisited(refOf(threadId)); + const unmount = h.registry.mount(atom); + const subscription = yield* Queue.take(h.subscriptions); + expect(subscription.threadId).toBe(threadId); + yield* Queue.offerAll(subscription.events, [ + event(threadId, deleted), + { kind: "synchronized" }, + ]); + yield* observeState(h.registry, atom, (state) => + deleted + ? state.status === "deleted" + : state.status === "live" && + Option.getOrNull(state.data)?.thread.title !== THREAD.thread.title, + ); + return { unmount, subscription }; + }); + + const visit = Effect.fn("TestThreadAtoms.visit")(function* ( + h: Harness, + threadId: ThreadId, + options?: { readonly deleted?: boolean; readonly secondary?: boolean }, + ) { + const { unmount } = yield* open(h, threadId, options); + unmount(); + yield* Queue.take(h.threadScopesClosed); + }); + + /** Reopens a left thread and reports whether it came from memory. */ + const reopen = Effect.fn("TestThreadAtoms.reopen")(function* (h: Harness, threadId: ThreadId) { + const diskLoads = h.counts().diskLoads; + const atom = atomOf(h, threadId); + h.rawAtoms.markThreadVisited(refOf(threadId)); + const unmount = h.registry.mount(atom); + const subscription = yield* Queue.take(h.subscriptions); + const warm = h.counts().diskLoads === diskLoads; + const state = yield* observeState(h.registry, atom, (value) => Option.isSome(value.data)); + unmount(); + yield* Queue.take(h.threadScopesClosed); + return { warm, state, afterSequence: subscription.afterSequence }; + }); + + it.effect("keeps left threads within the budget warm and reopens others from the cache", () => + Effect.gen(function* () { + const released: ThreadId[] = []; + const h = yield* makeHarness({ + projectionFor: projectionFor(), + leftThreads: budget(released), + }); + for (const threadId of LEFT.slice(0, 5)) yield* visit(h, threadId); + expect(released).toEqual([LEFT[0]]); + // Reopening refreshes a thread's place, so the next visit drops left-2 rather than left-1. + expect((yield* reopen(h, LEFT[1]!)).warm).toBe(true); + yield* visit(h, LEFT[5]!); + expect(released).toEqual([LEFT[0], LEFT[2]]); + const httpLoads = h.counts().httpLoads; + const reopened = yield* reopen(h, LEFT[0]!); + expect(reopened.warm).toBe(false); + const projection = Option.getOrThrow(reopened.state.data); + expect(projection.thread.title).toBe(`Visited ${LEFT[0]}`); + expect(projection.turnItems.map((item) => item.id)).toEqual([itemIdOf(LEFT[0]!)]); + expect(projection.visibleTurnItems.map((row) => row.sourceItemId)).toEqual([ + itemIdOf(LEFT[0]!), + ]); + expect(reopened.afterSequence).toBe(8); + expect(h.counts().httpLoads).toBe(httpLoads); + }), + ); + + it.effect("drops a heavy left thread and keeps the small ones warm", () => + Effect.gen(function* () { + const released: ThreadId[] = []; + const heavy = LEFT[4]!; + const h = yield* makeHarness({ + projectionFor: projectionFor({ + bytes: (threadId) => (threadId === heavy ? 40_000 : 1_000), + }), + leftThreads: budget(released), + }); + for (const threadId of LEFT.slice(0, 5)) yield* visit(h, threadId); + // Just left, so it does not count: nothing is dropped to make room for it. + expect(released).toEqual([]); + yield* visit(h, LEFT[5]!); + expect(released).toEqual([heavy]); + for (const threadId of LEFT.slice(0, 4)) { + expect((yield* reopen(h, threadId)).warm).toBe(true); + } + }), + ); + + it.effect("stays warm going back and forth between two threads over the budget", () => + Effect.gen(function* () { + const released: ThreadId[] = []; + const h = yield* makeHarness({ + projectionFor: projectionFor(), + leftThreads: budget(released, 0), + }); + yield* visit(h, LEFT[0]!); + yield* visit(h, LEFT[1]!); + expect(released).toEqual([LEFT[0]]); + yield* visit(h, LEFT[0]!); + expect((yield* reopen(h, LEFT[0]!)).warm).toBe(true); + expect((yield* reopen(h, LEFT[0]!)).warm).toBe(true); + }), + ); + + it.effect("does not let threads the user never navigated to crowd out left threads", () => + Effect.gen(function* () { + const released: ThreadId[] = []; + const h = yield* makeHarness({ + projectionFor: projectionFor(), + leftThreads: budget(released, 0), + }); + yield* visit(h, LEFT[0]!); + // A side chat, a fork source, a cited thread: all mount the same state atom. + for (const threadId of LEFT.slice(1)) yield* visit(h, threadId, { secondary: true }); + expect(released).toEqual([]); + expect((yield* reopen(h, LEFT[0]!)).warm).toBe(true); + expect((yield* reopen(h, LEFT[1]!)).warm).toBe(true); + }), + ); + + it.effect("keeps threads with unfinished runs outside the budget", () => + Effect.gen(function* () { + const released: ThreadId[] = []; + const running = new Set(LEFT.slice(0, 3)); + const h = yield* makeHarness({ + projectionFor: projectionFor({ running }), + leftThreads: budget(released, 15_000), + }); + for (const threadId of LEFT.slice(0, 4)) yield* visit(h, threadId); + // One settled thread fits, besides the one just left, whatever the running ones hold. + yield* visit(h, LEFT[4]!); + expect(released).toEqual([]); + yield* visit(h, LEFT[5]!); + expect(released).toEqual([LEFT[3]]); + const first = yield* reopen(h, LEFT[0]!); + expect(first.warm).toBe(true); + expect(Option.getOrThrow(first.state.data).runs[0]?.status).toBe("running"); + }), + ); + + describe("with a pending question", () => { + const QUESTION = { + id: RuntimeRequestId.make("left-question"), + nodeId: NodeId.make("left-question-node"), + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input" as const, + status: "pending" as const, + responseCapability: { type: "message" as const }, + createdAt: at, + resolvedAt: null, + }; + // As the server sends one that a run has pushed out of the window: after the window's items. + const withQuestion = (threadId: ThreadId): OrchestrationV2ThreadProjection => { + const projection = projectionFor()(threadId); + const { message: _message, ...notice } = itemsFor(threadId, 0).turnItems[0]!; + return { + ...projection, + runtimeRequests: [QUESTION], + turnItems: [ + ...projection.turnItems, + { + ...notice, + id: TurnItemId.make(`${threadId}-question`), + nodeId: QUESTION.nodeId, + ordinal: 0, + status: "running", + completedAt: null, + type: "user_input_request", + requestId: QUESTION.id, + questions: [{ id: "scope", header: "Scope", question: "Which scope?", options: [] }], + }, + ], + }; + }; + const questionsOf = (state: { + readonly data: Option.Option; + }) => + derivePendingThreadRequests(Option.getOrThrow(state.data)).userInputs.map( + (input) => input.requestId, + ); + + it.effect("stays outside the budget, even with no unfinished run", () => + Effect.gen(function* () { + const released: ThreadId[] = []; + const asking = LEFT[0]!; + const h = yield* makeHarness({ + projectionFor: (threadId) => + threadId === asking ? withQuestion(threadId) : projectionFor()(threadId), + leftThreads: budget(released, 0), + }); + for (const threadId of LEFT.slice(0, 3)) yield* visit(h, threadId); + expect(released).toEqual([LEFT[1]]); + const reopened = yield* reopen(h, asking); + expect(reopened.warm).toBe(true); + expect(questionsOf(reopened.state)).toEqual([QUESTION.id]); + }), + ); + + it.effect("is still answerable when the idle TTL released its thread", () => + Effect.gen(function* () { + const released: ThreadId[] = []; + const h = yield* makeHarness({ + projectionFor: withQuestion, + throughCacheRecord: true, + leftThreads: budget(released), + }); + const { unmount } = yield* open(h, LEFT[0]!); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + unmount(); + yield* Queue.take(h.threadScopesClosed); + for (let round = 0; round < 3; round++) { + vi.advanceTimersByTime(THREAD_SNAPSHOT_IDLE_TTL_MS + 1_000); + yield* Effect.yieldNow; + } + expect(released).toEqual([LEFT[0]]); + vi.useRealTimers(); + const httpLoads = h.counts().httpLoads; + const reopened = yield* reopen(h, LEFT[0]!); + expect(reopened.warm).toBe(false); + expect(reopened.afterSequence).toBe(8); + expect(h.counts().httpLoads).toBe(httpLoads); + expect(questionsOf(reopened.state)).toEqual([QUESTION.id]); + }), + ); + }); + + it.effect("keeps a deleted thread deleted after it is left", () => + Effect.gen(function* () { + const released: ThreadId[] = []; + const h = yield* makeHarness({ + projectionFor: projectionFor(), + leftThreads: budget(released, 0), + }); + yield* visit(h, LEFT[0]!, { deleted: true }); + for (const threadId of LEFT.slice(1)) yield* visit(h, threadId); + expect(released).not.toContain(LEFT[0]); + const diskLoads = h.counts().diskLoads; + const atom = atomOf(h, LEFT[0]!); + const unmount = h.registry.mount(atom); + yield* observeState(h.registry, atom, (state) => state.status === "deleted"); + expect(h.counts().diskLoads).toBe(diskLoads); + expect(yield* Queue.size(h.subscriptions)).toBe(0); + unmount(); + yield* Queue.take(h.threadScopesClosed); + }), + ); + + it.effect("leaves an open thread and the left threads alone when its environment restarts", () => + Effect.gen(function* () { + const released: ThreadId[] = []; + const h = yield* makeHarness({ + projectionFor: projectionFor(), + leftThreads: budget(released, 0), + }); + yield* visit(h, LEFT[0]!); + const { unmount } = yield* open(h, LEFT[1]!); + const before = currentThread(h.registry, atomOf(h, LEFT[1]!)); + const diskLoads = h.counts().diskLoads; + yield* h.restartEnvironment; + yield* Queue.take(h.threadScopesClosed); + const resumed = yield* Queue.take(h.subscriptions); + expect(resumed.threadId).toBe(LEFT[1]); + expect(resumed.afterSequence).toBe(8); + expect(h.counts().diskLoads).toBe(diskLoads); + expect(currentThread(h.registry, atomOf(h, LEFT[1]!))).toBe(before); + expect(released).toEqual([]); + unmount(); + yield* Queue.take(h.threadScopesClosed); + // Leaving it for real is what pushes the older thread out. + expect(released).toEqual([LEFT[0]]); + }), + ); + + it.effect("reopens a released expanded thread from its saved window", () => + Effect.gen(function* () { + const released: ThreadId[] = []; + const h = yield* makeHarness({ + snapshot: { ...SNAPSHOT, historyCursor: "older-1", hasMoreHistory: true }, + projectionFor: projectionFor(), + leftThreads: budget(released, 0), + }); + h.rawAtoms.markThreadVisited(h.ref); + const unmount = h.registry.mount(h.stateAtom); + const subscription = yield* Queue.take(h.subscriptions); + yield* Queue.offer(subscription.events, { kind: "synchronized" }); + yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live"); + const loading = yield* h.loadEarlier().pipe(Effect.forkScoped); + const request = yield* Queue.take(h.olderLoads); + const older = itemsFor(THREAD_ID, 10).visibleTurnItems[0]!; + const olderItem = { ...older.item, id: TurnItemId.make("older-item") }; + yield* Deferred.succeed(request.response, { + snapshotSequence: 7, + items: [{ ...older, sourceItemId: olderItem.id, item: olderItem }], + nextCursor: null, + hasMoreHistory: false, + }); + expect(yield* Fiber.join(loading)).toEqual({ _tag: "loaded" }); + yield* observeState(h.registry, h.stateAtom, (state) => state.history.expanded); + unmount(); + yield* Queue.take(h.threadScopesClosed); + + yield* visit(h, LEFT[0]!); + expect(released).toEqual([THREAD_ID]); + const httpLoads = h.counts().httpLoads; + const reopened = yield* reopen(h, THREAD_ID); + expect(reopened.warm).toBe(false); + expect(reopened.state.history).toMatchObject({ + expanded: false, + historyCursor: "older-1", + hasMoreHistory: true, + }); + expect( + Option.getOrThrow(reopened.state.data).visibleTurnItems.map((row) => row.sourceItemId), + ).toEqual([itemIdOf(THREAD_ID)]); + expect(h.counts().httpLoads).toBe(httpLoads); + }), + ); + + it.effect("keeps a thread in memory when its final save failed", () => + Effect.gen(function* () { + const released: ThreadId[] = []; + const h = yield* makeHarness({ + projectionFor: projectionFor(), + leftThreads: budget(released, 0), + beforeSave: (threadId) => + threadId === LEFT[0] + ? Effect.fail( + new Persistence.ConnectionPersistenceError({ + operation: "save-thread", + message: "quota", + }), + ) + : Effect.void, + }); + for (const threadId of LEFT.slice(0, 3)) yield* visit(h, threadId); + expect(released).toEqual([LEFT[1]]); + const kept = yield* reopen(h, LEFT[0]!); + expect(kept.warm).toBe(true); + expect(Option.getOrThrow(kept.state.data).thread.title).toBe(`Visited ${LEFT[0]}`); + }), + ); + + it.effect("releases a left thread only after its final save finished", () => + Effect.gen(function* () { + const released: ThreadId[] = []; + const saving = yield* Deferred.make(); + const gate = yield* Deferred.make(); + const h = yield* makeHarness({ + projectionFor: projectionFor(), + leftThreads: budget(released, 0), + beforeSave: (threadId) => + threadId === LEFT[0] + ? Deferred.succeed(saving, undefined).pipe(Effect.andThen(Deferred.await(gate))) + : Effect.void, + }); + const { unmount } = yield* open(h, LEFT[0]!); + unmount(); + yield* Deferred.await(saving); + // The stream's ensuring runs ahead of the scope's finalizers, so this one is already there. + yield* Queue.take(h.threadScopesClosed); + for (const threadId of LEFT.slice(1, 3)) yield* visit(h, threadId); + // left-0 is still saving, so it is not in the budget yet. + expect(released).toEqual([LEFT[1]]); + yield* Deferred.succeed(gate, undefined); + yield* Effect.yieldNow.pipe(Effect.repeat({ until: () => released.length === 2 })); + expect(released).toEqual([LEFT[1], LEFT[2]]); + yield* visit(h, LEFT[3]!); + expect(released).toEqual([LEFT[1], LEFT[2], LEFT[0]]); + const reopened = yield* reopen(h, LEFT[0]!); + expect(reopened.warm).toBe(false); + expect(Option.getOrThrow(reopened.state.data).thread.title).toBe(`Visited ${LEFT[0]}`); + }), + ); + + it.effect("reports a thread released by the idle TTL, visited or not", () => + Effect.gen(function* () { + const released: ThreadId[] = []; + const h = yield* makeHarness({ + projectionFor: projectionFor(), + leftThreads: budget(released), + }); + const { unmount } = yield* open(h, LEFT[0]!, { secondary: true }); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + unmount(); + yield* Queue.take(h.threadScopesClosed); + expect(released).toEqual([]); + // The derived atoms go first, on the registry's default TTL, and let go of the resume atom. + for (let round = 0; round < 3; round++) { + vi.advanceTimersByTime(THREAD_SNAPSHOT_IDLE_TTL_MS + 1_000); + yield* Effect.yieldNow; + } + expect(released).toEqual([LEFT[0]]); + }), + ); + + it.effect("keeps every left thread until the TTL without a budget", () => + Effect.gen(function* () { + const h = yield* makeHarness({ projectionFor: projectionFor() }); + for (const threadId of LEFT) yield* visit(h, threadId); + for (const threadId of LEFT) expect((yield* reopen(h, threadId)).warm).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 b99d04accaeea..5a8bf2a8488a3 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -535,6 +535,26 @@ describe("EnvironmentThreads", () => { }), ); + it.effect.each([ + ["has its card", true], + ["has no card", false], + ] as const)("counts as a saved copy to come back from only when it %s", ([, usable]) => + Effect.gen(function* () { + const stored = yield* roundTrip({ + ...PROJECTION_WITH_QUESTION, + turnItems: usable ? [QUESTION_CARD, newer] : [newer], + visibleTurnItems: usable ? [rowOf(QUESTION_CARD, 0), rowOf(newer, 1)] : [rowOf(newer, 0)], + }); + const resumeCache = { owner: undefined, snapshot: undefined, saved: true }; + yield* makeHarness({ + loadCached: Effect.succeedSome(stored.snapshot), + resumeCache, + completionMarker: true, + }); + expect(resumeCache.saved).toBe(usable); + }), + ); + it.effect("is fetched again when the record has no card for it", () => Effect.gen(function* () { const stored = yield* roundTrip({ diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 22c2a799d342f..b67e5334137eb 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -6,6 +6,7 @@ import { type OrchestrationV2ThreadProjection, type OrchestrationV2ThreadStreamItem, type OrchestrationV2TurnItem, + type ScopedThreadRef, type ThreadId as ThreadIdType, } from "@t3tools/contracts"; import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot"; @@ -35,7 +36,11 @@ import * as ManagedRelay from "../relay/managedRelay.ts"; import { request, subscribeDynamic } from "../rpc/client.ts"; import { parseThreadKey, threadKey } from "./entities.ts"; import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts"; -import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts"; +import { + estimateRetainedThreadBytes, + THREAD_SNAPSHOT_IDLE_TTL_MS, + UNFINISHED_RUN_STATUSES, +} from "./threadRetention.ts"; import { followStreamInEnvironment } from "./runtime.ts"; import * as ThreadHistoryController from "./threadHistoryController.ts"; import { fetchEnvironmentThreadHistoryPage } from "./threadHistoryHttp.ts"; @@ -157,6 +162,91 @@ interface ThreadResumeSnapshot { interface ThreadResumeCache { snapshot: ThreadResumeSnapshot | undefined; owner: object | undefined; + /** The family's budget for left threads, when the platform sets one. */ + readonly left?: LeftThreadBudget; + readonly key?: string; + /** Some thread-state atom has it open, even while an environment restart closes its scope. */ + mounted?: boolean; + /** The persisted cache has some copy of this thread, possibly an older bounded one. */ + saved?: boolean; + /** Its size when it was left. */ + bytes?: number; +} + +export interface LeftThreadRetention { + /** + * Past this estimated size, left threads give up their in-memory state, the largest first. The + * thread just left does not count, so going back and forth between two threads stays warm. + */ + readonly maxBytes: number; + /** A left thread's state is gone, through this budget or its idle TTL. */ + readonly onRelease?: (thread: ScopedThreadRef) => void; +} + +/** + * A deleted thread keeps its tombstone. A thread with unfinished runs and one whose latest save + * failed keep state the cache lacks. A thread with a pending question or approval waits for the user, + * so it stays ready to answer. A thread kept out of the cache on purpose, such as an expanded one, + * needs only some saved copy to come back from, and reloads its older pages on scroll. + */ +function isReleasable(cache: ThreadResumeCache, snapshot: ThreadResumeSnapshot): boolean { + const { state } = snapshot; + if (state.status === "deleted" || Option.isNone(state.data)) return false; + if (state.data.value.runs.some((run) => UNFINISHED_RUN_STATUSES.has(run.status))) return false; + if (pendingRuntimeRequestIds(state.data.value).length > 0) return false; + if (snapshot.persisted) return true; + return !shouldPersistThread(state.data.value, state.history) && cache.saved === true; +} + +class LeftThreadBudget { + /** Oldest first. Holds only threads the user navigated to, left, and can reopen from the cache. */ + private readonly order = new Set(); + private readonly visited = new Set(); + private readonly retention: LeftThreadRetention; + + constructor(retention: LeftThreadRetention) { + this.retention = retention; + } + + markVisited(key: string): void { + this.visited.add(key); + } + + open(cache: ThreadResumeCache): void { + cache.mounted = true; + this.order.delete(cache); + } + + /** Needs the atom unmounted and the scope closed, whichever comes last; the scope's close saves. */ + leave(cache: ThreadResumeCache): void { + if (cache.mounted === true || cache.owner !== undefined) return; + const snapshot = cache.snapshot; + if (snapshot === undefined || !this.visited.has(cache.key!)) return; + if (!isReleasable(cache, snapshot)) return; + cache.bytes = estimateRetainedThreadBytes(Option.getOrThrow(snapshot.state.data)); + this.order.delete(cache); + this.order.add(cache); + let total = -cache.bytes; + for (const entry of this.order) total += entry.bytes!; + while (total > this.retention.maxBytes) { + let largest: ThreadResumeCache | undefined; + for (const entry of this.order) { + // Among equals, the one left longest ago. + if (entry !== cache && entry.bytes! > (largest?.bytes ?? -1)) largest = entry; + } + if (largest === undefined) return; + total -= largest.bytes!; + largest.snapshot = undefined; + this.forget(largest); + } + } + + /** Reached by the budget above, and by the idle TTL for any thread. */ + forget(cache: ThreadResumeCache): void { + this.order.delete(cache); + this.visited.delete(cache.key!); + this.retention.onRelease?.(parseThreadKey(cache.key!)); + } } function matchesThreadSnapshot( @@ -228,6 +318,17 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make : candidate; const owner = {}; if (resumeCache) resumeCache.owner = owner; + if (resumeCache?.left) { + const { left } = resumeCache; + // Added first so it runs last, after the final cache save below. + yield* Effect.addFinalizer(() => + Effect.sync(() => { + if (resumeCache.owner !== owner) return; + resumeCache.owner = undefined; + left.leave(resumeCache); + }), + ); + } const loaded = retained === undefined ? yield* cache.loadThread(environmentId, threadId).pipe( @@ -249,6 +350,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make (snapshot) => !hasMissingPendingQuestionCard(snapshot.projection), ); const cachedThread = Option.map(cached, (snapshot) => snapshot.projection); + // A record discarded above is no copy to come back from. + if (resumeCache && Option.isSome(loaded)) resumeCache.saved = Option.isSome(cached); // Seed the resume cursor from the cached snapshot so a warm cache can catch up // via `afterSequence` instead of re-downloading the full thread body. const initialSequence = @@ -330,6 +433,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make Effect.tap(() => Effect.sync(() => { savedSnapshot = snapshot; + if (resumeCache?.owner === owner) resumeCache.saved = true; if ( !matchesThreadSnapshot( committed, @@ -1085,14 +1189,21 @@ export function createEnvironmentThreadStateAtoms( | R, E >, + options?: { + /** Without it, a left thread keeps its state for the whole idle TTL. */ + readonly leftThreads?: LeftThreadRetention; + }, ) { + const left = options?.leftThreads && new LeftThreadBudget(options.leftThreads); // Cache definitions must outlive collectible live-atom definitions. The // registry retains these nodes without retaining environment or RPC scopes. const resumeFamily = Atom.family((key: string) => - Atom.make((): ThreadResumeCache => ({ - snapshot: undefined, - owner: undefined, - })).pipe( + Atom.make((get): ThreadResumeCache => { + if (left === undefined) return { snapshot: undefined, owner: undefined }; + const cache: ThreadResumeCache = { snapshot: undefined, owner: undefined, left, key }; + get.addFinalizer(() => left.forget(cache)); + return cache; + }).pipe( Atom.setIdleTTL(THREAD_SNAPSHOT_IDLE_TTL_MS), Atom.withLabel(`environment-thread-resume:${key}`), ), @@ -1105,6 +1216,13 @@ export function createEnvironmentThreadStateAtoms( (get) => { get.mount(resumeAtom); const resume = get.once(resumeAtom); + if (left) { + left.open(resume); + get.addFinalizer(() => { + resume.mounted = false; + left.leave(resume); + }); + } const live = threadStateChanges(environmentId, threadId, resume); return resume.snapshot === undefined ? live @@ -1120,6 +1238,8 @@ export function createEnvironmentThreadStateAtoms( return { stateAtom: (environmentId: EnvironmentIdType, threadId: ThreadIdType) => family(threadKey({ environmentId, threadId })), + /** The user navigated to this thread: only such threads count toward the left-thread budget. */ + markThreadVisited: (ref: ScopedThreadRef) => left?.markVisited(threadKey(ref)), }; } From 36b2e23d9ae03a1ffac71f1f79c3a93897d000bb Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Thu, 5 Nov 2026 14:58:50 +0100 Subject: [PATCH 06/10] test(web): the queued-run test's threads mock gains the release hook ChatView.logic.ts now imports it from state/threads, and this mock replaces that whole module. --- apps/web/src/components/chat/useQueuedRunEdit.browser.test.tsx | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/web/src/components/chat/useQueuedRunEdit.browser.test.tsx b/apps/web/src/components/chat/useQueuedRunEdit.browser.test.tsx index cc6e04885c623..961582416da3f 100644 --- a/apps/web/src/components/chat/useQueuedRunEdit.browser.test.tsx +++ b/apps/web/src/components/chat/useQueuedRunEdit.browser.test.tsx @@ -64,6 +64,7 @@ vi.mock("../../state/threads", async () => { environmentThreadShells: { threadShellAtom: () => Atom.make(null), }, + onLeftThreadReleased: () => {}, }; }); From 223711901e220837b1bcce179f0b1f18a44622be Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Thu, 5 Nov 2026 15:07:50 +0100 Subject: [PATCH 07/10] test(client-runtime): the left-thread save test takes each release from a queue It spun on yieldNow, which the repo's testing rules ban. --- .../client-runtime/src/state/threads-atoms.test.ts | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/packages/client-runtime/src/state/threads-atoms.test.ts b/packages/client-runtime/src/state/threads-atoms.test.ts index ab4501a32f78a..446403513e380 100644 --- a/packages/client-runtime/src/state/threads-atoms.test.ts +++ b/packages/client-runtime/src/state/threads-atoms.test.ts @@ -973,11 +973,18 @@ describe("left threads", () => { it.effect("releases a left thread only after its final save finished", () => Effect.gen(function* () { const released: ThreadId[] = []; + const releases = yield* Queue.unbounded(); const saving = yield* Deferred.make(); const gate = yield* Deferred.make(); const h = yield* makeHarness({ projectionFor: projectionFor(), - leftThreads: budget(released, 0), + leftThreads: { + maxBytes: 0, + onRelease: (thread) => { + released.push(thread.threadId); + Queue.offerUnsafe(releases, undefined); + }, + }, beforeSave: (threadId) => threadId === LEFT[0] ? Deferred.succeed(saving, undefined).pipe(Effect.andThen(Deferred.await(gate))) @@ -992,7 +999,9 @@ describe("left threads", () => { // left-0 is still saving, so it is not in the budget yet. expect(released).toEqual([LEFT[1]]); yield* Deferred.succeed(gate, undefined); - yield* Effect.yieldNow.pipe(Effect.repeat({ until: () => released.length === 2 })); + // left-1's release is already queued; the finished save causes the next one. + yield* Queue.take(releases); + yield* Queue.take(releases); expect(released).toEqual([LEFT[1], LEFT[2]]); yield* visit(h, LEFT[3]!); expect(released).toEqual([LEFT[1], LEFT[2], LEFT[0]]); From 4953f591fa6be10cfbdcd88d410af58af0849ef5 Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Thu, 5 Nov 2026 15:09:11 +0100 Subject: [PATCH 08/10] perf(web): find lets go of the timeline it positioned once it has no match It kept the entries until its next successful reveal, so a thread find was last used in stayed in memory after it was released. --- .../chat/MessagesTimeline.browser.test.tsx | 49 +++++++++++++++++++ .../chat/useThreadFindNavigation.ts | 1 + 2 files changed, 50 insertions(+) diff --git a/apps/web/src/components/chat/MessagesTimeline.browser.test.tsx b/apps/web/src/components/chat/MessagesTimeline.browser.test.tsx index 775d32da13d27..ff316288b9731 100644 --- a/apps/web/src/components/chat/MessagesTimeline.browser.test.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.browser.test.tsx @@ -1221,6 +1221,55 @@ describe("MessagesTimeline thread find", () => { expect(top).toBeLessThan(bounds.bottom); }); }); + + it("lets go of a timeline once find has no match in it", async () => { + setup(); + const base = baseProps(); + await using screen = await renderElement( + , + ); + // In its own function, so nothing in this test's frame holds the entries after the switch. + const revealMatch = async () => { + const replies = Array.from({ length: 40 }, (_, index) => + message( + `${index === 5 ? "Needle here" : `Reply ${index}`}\n\n${"line\n\n".repeat(8)}`, + { role: "assistant" }, + `r-${index}`, + ), + ); + await screen.rerender( + , + ); + const viewport = base.listRef.current!.getScrollableNode()!; + await vi.waitFor(() => { + const row = screen.container.querySelector('[data-timeline-row-id="entry-r-5"]'); + expect(row).not.toBeNull(); + const bounds = viewport.getBoundingClientRect(); + const top = row!.getBoundingClientRect().top; + expect(top).toBeGreaterThanOrEqual(bounds.top); + expect(top).toBeLessThan(bounds.bottom); + }); + return new WeakRef(replies); + }; + const shown = await revealMatch(); + + // Twice, because React keeps the previous render's props until the next render. + for (const text of ["Another thread", "Another thread, updated"]) { + await screen.rerender(); + } + await cdp().send("HeapProfiler.collectGarbage"); + + expect(shown.deref()).toBeUndefined(); + }); }); describe("compact reasoning math", () => { diff --git a/apps/web/src/components/chat/useThreadFindNavigation.ts b/apps/web/src/components/chat/useThreadFindNavigation.ts index a00ab50f2f87e..419b3f724000a 100644 --- a/apps/web/src/components/chat/useThreadFindNavigation.ts +++ b/apps/web/src/components/chat/useThreadFindNavigation.ts @@ -86,6 +86,7 @@ function revealThreadFindMatch( state.settledMatch = null; state.revealedMatch = null; state.pagingPosition = null; + state.positionedEntries = null; return; } const list = listRef.current; From 5c53c4dda496b937b4a83f11df6e25194b8a6857 Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Thu, 5 Nov 2026 15:29:16 +0100 Subject: [PATCH 09/10] perf(web): the retention estimate charges a row that holds its own copy A version 5 or uncompacted record keeps a visible row whose copy differs from its item, and the estimate charged that row 64 bytes, so a large stale copy could take a thread past the budget unseen. --- .../src/state/threadRetention.test.ts | 10 ++++++++++ .../client-runtime/src/state/threadRetention.ts | 15 ++++++++------- 2 files changed, 18 insertions(+), 7 deletions(-) diff --git a/packages/client-runtime/src/state/threadRetention.test.ts b/packages/client-runtime/src/state/threadRetention.test.ts index 6d6aae3df6f54..507decffbb0df 100644 --- a/packages/client-runtime/src/state/threadRetention.test.ts +++ b/packages/client-runtime/src/state/threadRetention.test.ts @@ -45,4 +45,14 @@ describe("estimateRetainedThreadBytes", () => { } as unknown as OrchestrationV2ThreadProjection; expect(estimateRetainedThreadBytes(inherited)).toBeGreaterThan(1_000); }); + + it("charges a local row that holds its own copy of an item", () => { + const shared = withItems(1, 1_000); + const copied = { + ...shared, + visibleTurnItems: shared.visibleTurnItems.map((row) => ({ ...row, item: { ...row.item } })), + } as unknown as OrchestrationV2ThreadProjection; + const extra = estimateRetainedThreadBytes(copied) - estimateRetainedThreadBytes(shared); + expect(extra).toBeGreaterThan(1_000); + }); }); diff --git a/packages/client-runtime/src/state/threadRetention.ts b/packages/client-runtime/src/state/threadRetention.ts index 9b6d0f3de490f..4af6dc5d12e07 100644 --- a/packages/client-runtime/src/state/threadRetention.ts +++ b/packages/client-runtime/src/state/threadRetention.ts @@ -26,17 +26,18 @@ function retainedBytes(value: unknown): number { } /** - * A cheap estimate of what a thread keeps in the heap, from its items, inherited rows and message - * records, with strings counted by length. It walks one thread when the thread is left. Timeline - * rows and parsed markdown come on top, which matters only for small threads. + * A cheap estimate of what a thread keeps in the heap, from its items, rows holding an item of their + * own and message records, with strings counted by length. It walks one thread when the thread is + * left. Timeline rows and parsed markdown come on top, which matters only for small threads. */ export function estimateRetainedThreadBytes(projection: OrchestrationV2ThreadProjection): number { let bytes = retainedBytes(projection.turnItems) + retainedBytes(projection.messages); + const turnItems = new Set(projection.turnItems); for (const row of projection.visibleTurnItems) { - bytes += - row.sourceThreadId === projection.thread.id - ? OBJECT_OVERHEAD_BYTES - : retainedBytes(row.item) + OBJECT_OVERHEAD_BYTES; + // An inherited row, or a saved record's copy that differs from its item, holds its own. + bytes += turnItems.has(row.item) + ? OBJECT_OVERHEAD_BYTES + : retainedBytes(row.item) + OBJECT_OVERHEAD_BYTES; } return bytes; } From 431d40a8e078feb9ae3bee1a4915f732d7408c8b Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 10 Nov 2026 04:35:24 +0000 Subject: [PATCH 10/10] test(client): thread cache tests use the chunked encoder from main --- packages/client-runtime/src/state/threads-atoms.test.ts | 9 ++++++++- packages/client-runtime/src/state/threads-sync.test.ts | 9 ++++++++- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/packages/client-runtime/src/state/threads-atoms.test.ts b/packages/client-runtime/src/state/threads-atoms.test.ts index a1d0cd0feb395..aafc8ec8a4524 100644 --- a/packages/client-runtime/src/state/threads-atoms.test.ts +++ b/packages/client-runtime/src/state/threads-atoms.test.ts @@ -39,7 +39,7 @@ import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as Persistence from "../platform/persistence.ts"; import { decodeStoredThreadSnapshot, - encodeStoredThreadSnapshot, + makeStoredThreadSnapshotEncoder, } from "../platform/threadCacheEncoding.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; import type { RpcSession } from "../rpc/session.ts"; @@ -55,6 +55,13 @@ import { } from "./threads.ts"; import * as ThreadSnapshotLoader from "./threadSnapshotHttp.ts"; +const encodeCompactRecord = makeStoredThreadSnapshotEncoder({ + format: "compact", + yieldToHost: Effect.void, +}); +const encodeStoredThreadSnapshot = (stored: Parameters[0]) => + encodeCompactRecord(stored).pipe(Effect.map(Option.getOrThrow)); + const TARGET = new PrimaryConnectionTarget({ environmentId: EnvironmentId.make("environment-1"), label: "Test environment", diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 59ad72b0599f3..0bdb88a1c5a3c 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -48,7 +48,7 @@ import { } from "../platform/orchestrationCache.ts"; import { decodeStoredThreadSnapshot, - encodeStoredThreadSnapshot, + makeStoredThreadSnapshotEncoder, } from "../platform/threadCacheEncoding.ts"; import * as RpcSession from "../rpc/session.ts"; import { v2Now, v2Projection, v2ThreadId } from "./orchestrationV2TestFixtures.ts"; @@ -67,6 +67,13 @@ import { } from "./threads.ts"; import * as ThreadSnapshotLoader from "./threadSnapshotHttp.ts"; +const encodeCompactRecord = makeStoredThreadSnapshotEncoder({ + format: "compact", + yieldToHost: Effect.void, +}); +const encodeStoredThreadSnapshot = (stored: Parameters[0]) => + encodeCompactRecord(stored).pipe(Effect.map(Option.getOrThrow)); + const TARGET = new PrimaryConnectionTarget({ environmentId: EnvironmentId.make("environment-1"), label: "Test environment",