fix(server): scrolling back through a long thread no longer re-reads the whole thread on every page

apps/server/src/orchestration-v2/ProjectionStore.ts

@@ -110,6 +110,7 @@ import {
   THREAD_HISTORY_INLINE_WORK_ITEM_LIMIT,
   THREAD_HISTORY_TURN_SNAPSHOT_ROW_LIMIT,
 } from "./threadHistoryPaging.ts";
+import { makeThreadRevisionReader } from "./threadRevision.ts";
 
 export class ProjectionStoreApplyEventError extends Schema.TaggedError<ProjectionStoreApplyEventError>()(
   "ProjectionStoreApplyEventError",
@@ -5653,18 +5654,18 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
       readonly item: TimelineIndexItem;
       readonly synthetic?: OrchestrationV2TurnItem;
     };
+    type TimelineIndex = {
+      readonly records: ProjectionRecords<"runs" | "attempts">;
+      readonly local: ReadonlyArray<TimelineIndexRow>;
+      readonly visible: ReadonlyArray<TimelineIndexRow>;
+    };
     const readTimelineIndex: (
       threadId: ThreadId,
       seen: ReadonlySet<ThreadId>,
       includePersistentOutput?: boolean,
-    ) => Effect.Effect<
-      {
-        readonly records: ProjectionRecords<"runs" | "attempts">;
-        readonly local: ReadonlyArray<TimelineIndexRow>;
-        readonly visible: ReadonlyArray<TimelineIndexRow>;
-      },
-      ProjectionStoreV2Error
-    > = Effect.fn("ProjectionStore.readTimelineIndex")(
+    ) => Effect.Effect<TimelineIndex, ProjectionStoreV2Error> = Effect.fn(
+      "ProjectionStore.readTimelineIndex",
+    )(
       function* (threadId: ThreadId, seen: ReadonlySet<ThreadId>, includePersistentOutput = false) {
         const records = yield* getThreadRecords(threadId, ["runs", "attempts"]);
         const rows = yield* sql<{
@@ -5800,6 +5801,43 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
       readTimelineIndex(threadId, new Set()).pipe(Effect.map((index) => index.visible)),
     );
 
+    // Every history page needs the whole thread's index, so scrolling back through a long
+    // thread would rebuild it once per page. Entries are valid while the newest event of the
+    // thread and its fork sources is unchanged; a revision of 0 means no events to compare.
+    // A projection rebuild appends no events, so it needs a fresh store.
+    const readThreadRevision = yield* makeThreadRevisionReader;
+    const timelineIndexes = new Map<
+      string,
+      { readonly sequence: number; readonly index: TimelineIndex }
+    >();
+    const TIMELINE_INDEX_CACHE_CAPACITY = 4;
+    // The index holds attempt and persistent-output fields only when includePersistentOutput
+    // is set, so the two variants of a thread are cached separately.
+    const readCurrentTimelineIndex = Effect.fn("ProjectionStore.readCurrentTimelineIndex")(
+      function* (threadId: ThreadId, includePersistentOutput = false) {
+        const key = `${threadId}:${includePersistentOutput}`;
+        const sequence = yield* readThreadRevision([threadId]).pipe(
+          Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause })),
+        );
+        const cached = timelineIndexes.get(key);
+        if (cached !== undefined && cached.sequence === sequence && sequence > 0) {
+          timelineIndexes.delete(key);
+          timelineIndexes.set(key, cached);
+          return cached.index;
+        }
+        const index = yield* readTimelineIndex(threadId, new Set(), includePersistentOutput);
+        timelineIndexes.delete(key);
+        if (sequence > 0) {
+          timelineIndexes.set(key, { sequence, index });
+          if (timelineIndexes.size > TIMELINE_INDEX_CACHE_CAPACITY) {
+            const oldest = timelineIndexes.keys().next().value;
+            if (oldest !== undefined) timelineIndexes.delete(oldest);
+          }
+        }
+        return index;
+      },
+    );
+
     const getThreadHistoryPage: ProjectionStoreV2Shape["getThreadHistoryPage"] = (
       threadId,
       cursor,
@@ -5810,7 +5848,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
       sql
         .withTransaction(
           Effect.gen(function* () {
-            const index = yield* readTimelineIndex(threadId, new Set(), workMode !== undefined);
+            const index = yield* readCurrentTimelineIndex(threadId, workMode !== undefined);
             if (index.records.thread.deletedAt !== null)
               return yield* new ProjectionStoreThreadNotFoundError({ threadId });
             const revisions = yield* sql<{
@@ -5927,7 +5965,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
       sql
         .withTransaction(
           Effect.gen(function* () {
-            const index = yield* readTimelineIndex(threadId, new Set());
+            const index = yield* readCurrentTimelineIndex(threadId);
             const attachmentOwners =
               options.attachmentId === undefined
                 ? undefined

apps/server/src/orchestration-v2/ThreadFind.test.ts

@@ -822,4 +822,37 @@ describe("V2 history materialization", () => {
       assert.deepEqual(sizes, [1, 1, 75, 3]);
     }).pipe(Effect.provide(layerTest)),
   );
+
+  it.effect("keeps cached timeline reads current as the thread and its fork source change", () =>
+    Effect.gen(function* () {
+      const projection = yield* setup;
+      const parent = ThreadId.make("thread:parent");
+      const parentRun = RunId.make("run:parent");
+      yield* commit([thread(parent, projectId), run(parent, parentRun)]);
+      yield* putItems([{ ...item("p1", 1, "one", "assistant_message", parent), runId: parentRun }]);
+      const child = thread(threadId, projectId);
+      if (child.type !== "thread.created") return;
+      yield* commit([
+        {
+          ...child,
+          id: EventId.make("child:fork"),
+          type: "thread.metadata-updated",
+          payload: {
+            ...child.payload,
+            forkedFrom: { type: "run", threadId: parent, runId: parentRun },
+          },
+        },
+      ]);
+      yield* putItems([item("c1", 1, "local")]);
+      const totalItems = projection
+        .getTimelinePage(threadId, { view: "activity", limit: 100 })
+        .pipe(Effect.map((page) => page.totalItems));
+      const initial = yield* totalItems;
+      assert.equal(yield* totalItems, initial);
+      yield* putItems([item("c2", 2, "local again")]);
+      assert.equal(yield* totalItems, initial + 1);
+      yield* putItems([{ ...item("p2", 2, "two", "assistant_message", parent), runId: parentRun }]);
+      assert.equal(yield* totalItems, initial + 2);
+    }).pipe(Effect.provide(layerTest)),
+  );
 });

apps/server/src/orchestration-v2/ThreadFind.ts

@@ -21,6 +21,7 @@ import {
   ProjectionStoreThreadNotFoundError,
   type ProjectionStoreV2Error,
 } from "./ProjectionStore.ts";
+import { makeThreadRevisionReader } from "./threadRevision.ts";
 
 interface FindRow {
   readonly position: number;
@@ -183,7 +184,6 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* (
       }),
     ),
   );
-  const encodeThreadIds = Schema.encodeEffect(Schema.fromJsonString(Schema.Array(ThreadId)));
   const encodeSources = Schema.encodeEffect(
     Schema.fromJsonString(Schema.Array(Schema.Struct({ threadId: ThreadId, id: TurnItemId }))),
   );
@@ -252,28 +252,7 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* (
       sourceThreadIds: readonly ThreadId[];
     }
   >();
-  const readRevision = Effect.fn("ThreadFind.readRevision")(function* (
-    threadIds: readonly ThreadId[],
-  ) {
-    const sources = yield* encodeThreadIds(threadIds);
-    const revision = yield* sql<{ sequence: number }>`
-      WITH RECURSIVE source_threads(thread_id) AS (
-        SELECT value FROM json_each(${sources})
-        UNION
-        SELECT json_extract(t.payload_json, '$.forkedFrom.threadId')
-        FROM orchestration_v2_projection_threads t
-        JOIN source_threads s ON s.thread_id = t.thread_id
-        WHERE json_extract(t.payload_json, '$.forkedFrom.type') = 'run'
-      )
-      SELECT COALESCE(MAX(sequence), 0) AS sequence FROM orchestration_events
-      -- The planner otherwise walks every V2 event backwards from the newest,
-      -- which takes seconds for a thread that has been idle a long time.
-      INDEXED BY idx_orch_events_stream_sequence
-      WHERE application_event_version = 2 AND aggregate_kind = 'thread'
-        AND stream_id IN (SELECT thread_id FROM source_threads)
-    `;
-    return revision[0]?.sequence ?? 0;
-  });
+  const readRevision = yield* makeThreadRevisionReader;
   const snapshot = Effect.fn("ThreadFind.snapshot")(
     function* (input: OrchestrationV2SearchThreadInput) {
       const { threadId } = input;

apps/server/src/orchestration-v2/threadRevision.ts

@@ -0,0 +1,33 @@
+import { ThreadId } from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+import * as Schema from "effect/Schema";
+import * as SqlClient from "effect/sql/SqlClient";
+
+/**
+ * Latest event sequence across the given threads and every thread they fork
+ * from. Cached reads compare it with the value they were built at.
+ */
+export const makeThreadRevisionReader = Effect.gen(function* () {
+  const sql = yield* SqlClient.SqlClient;
+  const encodeThreadIds = Schema.encodeEffect(Schema.fromJsonString(Schema.Array(ThreadId)));
+  return Effect.fn("ThreadFind.readRevision")(function* (threadIds: readonly ThreadId[]) {
+    const sources = yield* encodeThreadIds(threadIds);
+    const revision = yield* sql<{ sequence: number }>`
+      WITH RECURSIVE source_threads(thread_id) AS (
+        SELECT value FROM json_each(${sources})
+        UNION
+        SELECT json_extract(t.payload_json, '$.forkedFrom.threadId')
+        FROM orchestration_v2_projection_threads t
+        JOIN source_threads s ON s.thread_id = t.thread_id
+        WHERE json_extract(t.payload_json, '$.forkedFrom.type') = 'run'
+      )
+      SELECT COALESCE(MAX(sequence), 0) AS sequence FROM orchestration_events
+      -- The planner otherwise walks every V2 event backwards from the newest,
+      -- which takes seconds for a thread that has been idle a long time.
+      INDEXED BY idx_orch_events_stream_sequence
+      WHERE application_event_version = 2 AND aggregate_kind = 'thread'
+        AND stream_id IN (SELECT thread_id FROM source_threads)
+    `;
+    return revision[0]?.sequence ?? 0;
+  });
+});

apps/server/src/terminal/OutputProtocol.test.ts

@@ -80,6 +80,78 @@ describe("terminal output window", () => {
       assert.equal((yield* Queue.take(responses))._tag, "Exit");
     }).pipe(Effect.provide(RpcSerialization.layerJson), Effect.scoped),
   );
+
+  const bigChunkEvents = [
+    {
+      name: "error messages",
+      event: {
+        threadId: "thread-1",
+        terminalId: "default",
+        type: "error" as const,
+        message: "e".repeat(64 * 1024),
+      },
+    },
+    {
+      name: "thread ids",
+      event: {
+        threadId: "t".repeat(64 * 1024),
+        terminalId: "default",
+        type: "output" as const,
+        data: "x",
+      },
+    },
+  ];
+
+  it.effect.each(bigChunkEvents)("charges $name against the pending-byte limit", ({ event }) =>
+    Effect.gen(function* () {
+      const tag = WS_METHODS.terminalAttach;
+      const group = RpcGroup.make(
+        Rpc.make(tag, { success: TerminalAttachStreamEventSchema, stream: true }),
+      );
+      const output =
+        yield* Queue.unbounded<Schema.Schema.Type<typeof TerminalAttachStreamEventSchema>>();
+      const responses = yield* Queue.unbounded<RpcMessage.FromServerEncoded>();
+      const receive = yield* Deferred.make<Parameters<RpcServer.Protocol["Service"]["run"]>[0]>();
+      const protocol = yield* RpcServer.Protocol.make((write) =>
+        Effect.gen(function* () {
+          yield* Deferred.succeed(receive, write);
+          const serialization = yield* RpcSerialization.RpcSerialization;
+          return {
+            disconnects: yield* Queue.unbounded<number>(),
+            send: (_clientId, response) => Queue.offer(responses, response),
+            end: () => Effect.void,
+            clientIds: Effect.succeed(new Set([0])),
+            initialMessage: Effect.succeedNone,
+            supportsAck: true,
+            supportsTransferables: false,
+            supportsSpanPropagation: false,
+            supportsNotifications: true,
+            codecFor: serialization.codecFor,
+          };
+        }),
+      );
+      yield* RpcServer.make(group).pipe(
+        Effect.provide(group.toLayerHandler(tag, () => Stream.fromQueue(output))),
+        Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)),
+        Effect.forkScoped,
+      );
+      const write = yield* Deferred.await(receive);
+      yield* write(0, { _tag: "Request", id: "1", tag, payload: null, headers: [] });
+
+      yield* Queue.offer(output, event);
+      yield* TestClock.adjust(0);
+      assert.equal(yield* Queue.size(responses), 1);
+      yield* Queue.take(responses);
+      // One 64 KiB field fills the byte window on its own.
+      yield* Queue.offer(output, event);
+      yield* TestClock.adjust(0);
+      assert.equal(yield* Queue.size(responses), 0);
+      yield* write(0, { _tag: "Ack", requestId: "1" });
+      assert.equal((yield* Queue.take(responses))._tag, "Chunk");
+      yield* write(0, { _tag: "Interrupt", requestId: "1" });
+      assert.equal((yield* Queue.take(responses))._tag, "Exit");
+    }).pipe(Effect.provide(RpcSerialization.layerJson), Effect.scoped),
+  );
 });
 
 const output = (data: string): TerminalAttachStreamEvent => ({

apps/server/src/terminal/OutputProtocol.ts

@@ -8,10 +8,65 @@ import type { RpcServer } from "effect/rpc";
 
 const MAX_PENDING_CHUNKS = 8;
 const MAX_PENDING_BYTES = 64 * 1024;
+// Charge an envelope so tiny events cannot bypass the byte budget.
+const EVENT_OVERHEAD = 128;
 const isFull = (sizes: number[]) =>
   sizes.length >= MAX_PENDING_CHUNKS ||
   sizes.reduce((total, size) => total + size, 0) >= MAX_PENDING_BYTES;
 
+const byteLengthOf = (value: unknown) => (typeof value === "string" ? Buffer.byteLength(value) : 0);
+
+// Charges every variable-size field a terminal event can carry, so the byte
+// budgets below cannot be bypassed by a large payload in an uncharged field.
+const queuedSize = (event: TerminalEvent | TerminalAttachStreamEvent) =>
+  EVENT_OVERHEAD +
+  ("threadId" in event ? byteLengthOf(event.threadId) + byteLengthOf(event.terminalId) : 0) +
+  (event.type === "output"
+    ? byteLengthOf(event.data)
+    : event.type === "error"
+      ? byteLengthOf(event.message)
+      : event.type === "activity"
+        ? byteLengthOf(event.label)
+        : "snapshot" in event
+          ? byteLengthOf(event.snapshot.threadId) +
+            byteLengthOf(event.snapshot.terminalId) +
+            byteLengthOf(event.snapshot.history) +
+            byteLengthOf(event.snapshot.cwd) +
+            byteLengthOf(event.snapshot.worktreePath) +
+            byteLengthOf(event.snapshot.label)
+          : 0);
+
+const TERMINAL_EVENT_TYPES = new Set([
+  "started",
+  "output",
+  "exited",
+  "closed",
+  "error",
+  "cleared",
+  "restarted",
+  "activity",
+  "snapshot",
+]);
+
+// Windowed chunks carry terminal events, so charge them the way the output
+// queue below does instead of serializing the frame a second time just to
+// measure it. Only the event shapes queuedSize measures take the fast path;
+// anything else keeps the exact JSON measurement.
+const chunkSize = (values: ReadonlyArray<unknown>) =>
+  values.reduce<number>((total, value) => {
+    if (typeof value === "string") return total + EVENT_OVERHEAD + Buffer.byteLength(value);
+    if (
+      typeof value === "object" &&
+      value !== null &&
+      "type" in value &&
+      typeof value.type === "string" &&
+      TERMINAL_EVENT_TYPES.has(value.type)
+    ) {
+      return total + queuedSize(value as TerminalEvent | TerminalAttachStreamEvent);
+    }
+    return total + Buffer.byteLength(JSON.stringify(value) ?? "");
+  }, EVENT_OVERHEAD);
+
 export function withTerminalOutputWindow(
   protocol: RpcServer.Protocol["Service"],
 ): RpcServer.Protocol["Service"] {
@@ -50,9 +105,7 @@ export function withTerminalOutputWindow(
         } else if (response._tag === "Chunk") {
           const window = windows.get(`${clientId}:${response.requestId}`);
           if (window) {
-            // @effect-diagnostics-next-line preferSchemaOverJson:off
-            const size = Buffer.byteLength(JSON.stringify(response));
-            window.push(size);
+            window.push(chunkSize(response.values));
             if (!isFull(window)) {
               return send.pipe(
                 Effect.andThen(() =>
@@ -71,8 +124,6 @@ export function withTerminalOutputWindow(
 }
 
 const MAX_QUEUED_OUTPUT = 8 * 1024 * 1024;
-// Charge an envelope so tiny events cannot bypass the byte budget.
-const EVENT_OVERHEAD = 128;
 
 type Listener<A> = (event: A) => Effect.Effect<void>;
 // Acquisition must register cleanup synchronously with installing the listener.
@@ -81,14 +132,6 @@ type Subscribe<A, E> = (
   registerCleanup: (stop: () => void) => void,
 ) => Effect.Effect<() => void, E>;
 
-const queuedSize = (event: TerminalEvent | TerminalAttachStreamEvent) =>
-  EVENT_OVERHEAD +
-  (event.type === "output"
-    ? Buffer.byteLength(event.data)
-    : "snapshot" in event
-      ? Buffer.byteLength(event.snapshot.history)
-      : 0);
-
 export function terminalOutputStream<A extends TerminalEvent | TerminalAttachStreamEvent, E>(
   subscribe: Subscribe<A, E>,
   resync?: Subscribe<A, E>,