fix(server): reuse timeline indexes while paging long threads

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

@@ -2929,6 +2929,16 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
       const now = yield* DateTime.now;
       const nowIso = DateTime.formatIso(now);
       const threadId = ThreadId.make("thread:bounded-sql-history");
+      // The raw SQL edits below stand in for committed writes, which always append a thread event.
+      let revision = 0;
+      const historyPage = (...args: Parameters<typeof projectionStore.getThreadHistoryPage>) =>
+        sql`INSERT INTO orchestration_events (
+          event_id, aggregate_kind, stream_id, stream_version, event_type,
+          occurred_at, actor_kind, payload_json, metadata_json, application_event_version
+        ) VALUES (
+          ${`event:bounded-sql-history:revision:${++revision}`}, 'thread', ${threadId}, ${revision},
+          'turn-item.updated', ${nowIso}, 'user', '{}', '{}', 2
+        )`.pipe(Effect.andThen(projectionStore.getThreadHistoryPage(...args)));
       yield* projectionStore.apply({
         id: EventId.make("event:bounded-sql-history:thread"),
         type: "thread.created",
@@ -3152,7 +3162,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
         sourceItemId: TurnItemId.make("turn-item:bounded-sql-history:71"),
         position: 70,
       });
-      const bytePage = yield* projectionStore.getThreadHistoryPage(threadId, byteCursor);
+      const bytePage = yield* historyPage(threadId, byteCursor);
       assert.isAbove(bytePage.items.length, 0);
       assert.isBelow(bytePage.items.length, 70);
       assert.isAtMost(
@@ -3163,7 +3173,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
       let byteNext = bytePage.nextCursor;
       for (let pages = 0; byteNext !== null; pages += 1) {
         assert.isBelow(pages, 70);
-        const page = yield* projectionStore.getThreadHistoryPage(threadId, byteNext);
+        const page = yield* historyPage(threadId, byteNext);
         assert.isAbove(page.items.length, 0);
         byteIds.unshift(...page.items.map((row) => String(row.sourceItemId)));
         byteNext = page.nextCursor;
@@ -3269,7 +3279,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
           '$.attachments', json('[]'))
         WHERE thread_id = ${threadId} AND ordinal BETWEEN 600 AND 624
       `;
-      const targetedPage = yield* projectionStore.getThreadHistoryPage(
+      const targetedPage = yield* historyPage(
         threadId,
         encodeThreadHistoryCursor({
           snapshotSequence: 0,
@@ -3299,7 +3309,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
             '$.type', CASE ordinal WHEN 500 THEN 'run_interrupt_request' ELSE 'run_interrupt_result' END)
         WHERE thread_id = ${threadId} AND ordinal IN (500, 580)
       `;
-      const interruptPage = yield* projectionStore.getThreadHistoryPage(
+      const interruptPage = yield* historyPage(
         threadId,
         encodeThreadHistoryCursor({
           snapshotSequence: 0,
@@ -3446,7 +3456,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
       let workCursor: string | null = anchor.item.deferredWork!.cursor;
       for (let pages = 0; workCursor !== null; pages += 1) {
         assert.isBelow(pages, 20);
-        const page: OrchestrationV2ThreadHistoryPage = yield* projectionStore.getThreadHistoryPage(
+        const page: OrchestrationV2ThreadHistoryPage = yield* historyPage(
           threadId,
           workCursor,
           anchor.sourceItemId,
@@ -3475,7 +3485,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
           window.projection.visibleTurnItems.find((row) => row.item.deferredWork)?.item.ordinal,
           3,
         );
-        const page = yield* projectionStore.getThreadHistoryPage(
+        const page = yield* historyPage(
           threadId,
           encodeThreadHistoryCursor({
             snapshotSequence: 0,
@@ -3532,14 +3542,13 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
         const loaded = new Set<TurnItemId>();
         for (let pages = 0; cursor !== null; pages += 1) {
           assert.isBelow(pages, 20);
-          const page: OrchestrationV2ThreadHistoryPage =
-            yield* projectionStore.getThreadHistoryPage(
-              threadId,
-              cursor,
-              oldWork!.sourceItemId,
-              false,
-              "work",
-            );
+          const page: OrchestrationV2ThreadHistoryPage = yield* historyPage(
+            threadId,
+            cursor,
+            oldWork!.sourceItemId,
+            false,
+            "work",
+          );
           for (const row of page.items) loaded.add(row.sourceItemId);
           cursor = page.nextCursor;
         }
@@ -4470,6 +4479,276 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
     }),
   );
 
+  it.effect("pages a whole thread without re-reading its full timeline for every page", () =>
+    Effect.gen(function* () {
+      const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
+      const sql = yield* SqlClient.SqlClient;
+      const now = yield* DateTime.now;
+      const nowIso = DateTime.formatIso(now);
+      const threadId = ThreadId.make("thread:paged-scrollback");
+      const runId = RunId.make("run:paged-scrollback");
+      const itemCount = 600;
+      yield* projectionStore.apply({
+        id: EventId.make("event:paged-scrollback:thread"),
+        type: "thread.created",
+        threadId,
+        occurredAt: now,
+        payload: {
+          createdBy: "user",
+          creationSource: "web",
+          id: threadId,
+          projectId: ProjectId.make("project:paged-scrollback"),
+          title: "Paged scrollback",
+          providerInstanceId,
+          modelSelection,
+          runtimeMode: "full-access",
+          interactionMode: "default",
+          branch: null,
+          worktreePath: null,
+          activeProviderThreadId: null,
+          lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
+          forkedFrom: null,
+          createdAt: now,
+          updatedAt: now,
+          archivedAt: null,
+          settledOverride: null,
+          settledAt: null,
+          lastVisitedAt: null,
+          deletedAt: null,
+        },
+      });
+      yield* sql`
+        INSERT INTO orchestration_v2_projection_runs (
+          run_id, thread_id, ordinal, provider, provider_thread_id, status,
+          requested_at, completed_at, payload_json
+        ) VALUES (
+          ${runId}, ${threadId}, 1, 'codex', NULL, 'completed', ${nowIso}, ${nowIso},
+          ${encodeUnknownJsonString({
+            id: runId,
+            threadId,
+            ordinal: 1,
+            providerInstanceId,
+            modelSelection,
+            providerThreadId: null,
+            userMessageId: "message:paged-scrollback",
+            rootNodeId: "node:paged-scrollback",
+            activeAttemptId: null,
+            status: "completed",
+            requestedAt: nowIso,
+            startedAt: nowIso,
+            completedAt: nowIso,
+            checkpointId: null,
+            contextHandoffId: null,
+          })}
+        )
+      `;
+      for (let ordinal = 0; ordinal < itemCount; ordinal += 1) {
+        const id = `turn-item:paged-scrollback:${ordinal}`;
+        const message = ordinal % 9 === 0;
+        yield* sql`
+          INSERT INTO orchestration_v2_projection_turn_items (
+            turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
+            parent_item_id, ordinal, type, status, updated_at, payload_json
+          ) VALUES (
+            ${id}, ${threadId}, ${runId}, NULL, NULL, NULL, NULL, ${ordinal},
+            ${message ? "user_message" : "command_execution"}, 'completed', ${nowIso},
+            ${encodeUnknownJsonString({
+              id,
+              threadId,
+              runId,
+              nodeId: null,
+              providerThreadId: null,
+              providerTurnId: null,
+              nativeItemRef: null,
+              parentItemId: null,
+              ordinal,
+              status: "completed",
+              title: `item ${ordinal}`,
+              startedAt: nowIso,
+              completedAt: nowIso,
+              updatedAt: nowIso,
+              ...(message
+                ? {
+                    type: "user_message",
+                    createdBy: "user",
+                    creationSource: "web",
+                    messageId: `message:paged-scrollback:${ordinal}`,
+                    inputIntent: "turn_start",
+                    text: `Turn ${ordinal}`,
+                    attachments: [],
+                  }
+                : { type: "command_execution", input: "echo", output: "ok", exitCode: 0 }),
+            })}
+          )
+        `;
+      }
+      // An unbounded read of every turn item in the thread costs O(thread) per page, so a full
+      // scrollback that repeats it on each page is quadratic in thread length.
+      const wholeThreadReads = (statements: ReadonlyArray<string>) =>
+        statements.filter(
+          (statement) =>
+            /FROM orchestration_v2_projection_turn_items(\s+AS item)?\s+WHERE (item\.)?thread_id = \?\s*(ORDER BY|$)/.test(
+              statement,
+            ) && !/\bLIMIT\b/.test(statement),
+        ).length;
+      const snapshot = yield* projectionStore.getThreadSnapshot(threadId);
+
+      const history = traceSqlStatements();
+      let cursor: string | null = encodeThreadHistoryCursor({
+        snapshotSequence: snapshot.snapshotSequence,
+        sourceThreadId: threadId,
+        sourceItemId: "absent",
+        position: itemCount,
+      });
+      const historyIds: Array<string> = [];
+      let historyPages = 0;
+      while (cursor !== null) {
+        const page: OrchestrationV2ThreadHistoryPage = yield* projectionStore
+          .getThreadHistoryPage(threadId, cursor)
+          .pipe(Effect.withTracer(history.tracer));
+        historyIds.push(...page.items.map((row) => String(row.sourceItemId)));
+        historyPages += 1;
+        cursor = page.hasMoreHistory ? page.nextCursor : null;
+      }
+      assert.strictEqual(new Set(historyIds).size, itemCount);
+      assert.isAtLeast(historyPages, 4);
+
+      const timeline = traceSqlStatements();
+      const timelineIds: Array<string> = [];
+      let afterPosition = -1;
+      let timelinePages = 0;
+      for (let hasMore = true; hasMore;) {
+        const page = yield* projectionStore
+          .getTimelinePage(threadId, { view: "activity", afterPosition, limit: 100 })
+          .pipe(Effect.withTracer(timeline.tracer));
+        timelineIds.push(...page.items.map((row) => String(row.sourceItemId)));
+        afterPosition = page.items.at(-1)?.position ?? afterPosition;
+        timelinePages += 1;
+        hasMore = page.hasMore;
+      }
+      assert.strictEqual(new Set(timelineIds).size, itemCount);
+      assert.isAtLeast(timelinePages, 4);
+
+      assert.isAtMost(
+        wholeThreadReads(history.statements),
+        1,
+        `getThreadHistoryPage read the whole thread on each of ${historyPages} pages`,
+      );
+      assert.isAtMost(
+        wholeThreadReads(timeline.statements),
+        1,
+        `getTimelinePage read the whole thread on each of ${timelinePages} pages`,
+      );
+
+      // A write to the thread a fork inherits from must reach the fork's next page.
+      const forkThreadId = ThreadId.make("thread:paged-scrollback:fork");
+      yield* projectionStore.apply({
+        id: EventId.make("event:paged-scrollback:fork"),
+        type: "thread.created",
+        threadId: forkThreadId,
+        occurredAt: now,
+        payload: {
+          createdBy: "user",
+          creationSource: "web",
+          id: forkThreadId,
+          projectId: ProjectId.make("project:paged-scrollback"),
+          title: "Paged scrollback fork",
+          providerInstanceId,
+          modelSelection,
+          runtimeMode: "full-access",
+          interactionMode: "default",
+          branch: null,
+          worktreePath: null,
+          activeProviderThreadId: null,
+          lineage: {
+            parentThreadId: threadId,
+            relationshipToParent: "fork",
+            rootThreadId: threadId,
+          },
+          forkedFrom: { type: "run", threadId, runId },
+          createdAt: now,
+          updatedAt: now,
+          archivedAt: null,
+          settledOverride: null,
+          settledAt: null,
+          lastVisitedAt: null,
+          deletedAt: null,
+        },
+      });
+      const forkPage = () =>
+        projectionStore.getTimelinePage(forkThreadId, { view: "activity", limit: 1 });
+      const forkTotal = (yield* forkPage()).totalItems;
+      assert.strictEqual((yield* forkPage()).totalItems, forkTotal);
+      yield* projectionStore.apply({
+        id: EventId.make("event:paged-scrollback:late-item"),
+        type: "turn-item.updated",
+        threadId,
+        runId,
+        driver,
+        occurredAt: now,
+        payload: {
+          id: TurnItemId.make("turn-item:paged-scrollback:late"),
+          threadId,
+          runId,
+          nodeId: null,
+          providerThreadId: null,
+          providerTurnId: null,
+          nativeItemRef: null,
+          parentItemId: null,
+          ordinal: itemCount,
+          status: "completed",
+          title: "late",
+          startedAt: now,
+          completedAt: now,
+          updatedAt: now,
+          type: "command_execution",
+          input: "echo",
+        },
+      });
+      assert.strictEqual((yield* forkPage()).totalItems, forkTotal + 1);
+
+      // Another store instance's write reaches this one only through the committed thread event.
+      const directId = "turn-item:paged-scrollback:direct";
+      yield* sql`
+        INSERT INTO orchestration_v2_projection_turn_items (
+          turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
+          parent_item_id, ordinal, type, status, updated_at, payload_json
+        ) VALUES (
+          ${directId}, ${threadId}, ${runId}, NULL, NULL, NULL, NULL, ${itemCount + 1},
+          'command_execution', 'completed', ${nowIso},
+          ${encodeUnknownJsonString({
+            id: directId,
+            threadId,
+            runId,
+            nodeId: null,
+            providerThreadId: null,
+            providerTurnId: null,
+            nativeItemRef: null,
+            parentItemId: null,
+            ordinal: itemCount + 1,
+            status: "completed",
+            title: "direct",
+            startedAt: nowIso,
+            completedAt: nowIso,
+            updatedAt: nowIso,
+            type: "command_execution",
+            input: "echo",
+          })}
+        )
+      `;
+      yield* sql`
+        INSERT INTO orchestration_events (
+          event_id, aggregate_kind, stream_id, stream_version, event_type,
+          occurred_at, actor_kind, payload_json, metadata_json, application_event_version
+        ) VALUES (
+          'event:paged-scrollback:direct', 'thread', ${threadId}, 1,
+          'turn-item.updated', ${nowIso}, 'user', '{}', '{}', 2
+        )
+      `;
+      assert.strictEqual((yield* forkPage()).totalItems, forkTotal + 2);
+    }),
+  );
+
   it.effect("scopes actionable provider state to the requested bounded thread", () =>
     Effect.gen(function* () {
       const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
@@ -4781,6 +5060,8 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
           number: 42,
           url: "https://github.com/acme/project/pull/42",
         })})) WHERE thread_id = ${readThreadId}`;
+      const [latest] = yield* sql<{ readonly sequence: number }>`
+        SELECT COALESCE(MAX(sequence), 0) AS sequence FROM orchestration_events`;
       const trace = traceSqlStatements();
       const discovery = yield* projectionStore
         .getPullRequestSnapshot({})
@@ -4791,7 +5072,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
           .filter((id) => id === readThreadId || id === laterThreadId),
         [readThreadId],
       );
-      assert.strictEqual(discovery.snapshotSequence, 0);
+      assert.strictEqual(discovery.snapshotSequence, latest?.sequence);
       assert.strictEqual(trace.queryTransactions.length, 2);
       assert.isNotNull(trace.queryTransactions[0]);
       assert.strictEqual(new Set(trace.queryTransactions).size, 1);