fix(server): enforce history page limits at turn boundaries

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

@@ -9,6 +9,7 @@ import {
   MessageId,
   type ModelSelection,
   type OrchestrationV2ProviderThread,
+  type OrchestrationV2ThreadHistoryPage,
   type OrchestrationV2ThreadShellSnapshot,
   NodeId,
   ProjectId,
@@ -1091,7 +1092,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
     }),
   );
 
-  it.effect("pages complete user turns through SQL regardless of tool count or payload size", () =>
+  it.effect("pages oversized user turns individually through SQL without splitting them", () =>
     Effect.gen(function* () {
       const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
       const sql = yield* SqlClient.SqlClient;
@@ -1236,11 +1237,11 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
         projection: initial.projection,
         snapshotSequence: 0,
       });
-      assert.lengthOf(bounded.projection.visibleTurnItems, 10 * 102);
-      assert.strictEqual(bounded.projection.visibleTurnItems[0]?.sourceItemId, allIds[35 * 102]);
+      assert.lengthOf(bounded.projection.visibleTurnItems, 102);
+      assert.strictEqual(bounded.projection.visibleTurnItems[0]?.sourceItemId, allIds[44 * 102]);
       const loaded = bounded.projection.visibleTurnItems.map((row) => String(row.sourceItemId));
       let cursor = bounded.historyCursor;
-      for (const turns of [20, 15]) {
+      for (let turn = 44; turn >= 1; turn -= 1) {
         assert.isNotNull(cursor);
         const anchor = decodeThreadHistoryCursor(cursor!);
         const snapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, {
@@ -1254,7 +1255,8 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
           cursor: cursor!,
           snapshotSequence: 0,
         });
-        assert.lengthOf(page.items, turns * 102);
+        assert.lengthOf(page.items, 102);
+        assert.strictEqual(page.items[0]?.sourceItemId, allIds[(turn - 1) * 102]);
         loaded.unshift(...page.items.map((row) => String(row.sourceItemId)));
         cursor = page.nextCursor;
       }
@@ -1423,11 +1425,12 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
         projection: initial.projection,
         snapshotSequence: 0,
       });
-      assert.strictEqual(bounded.projection.visibleTurnItems[0]?.sourceItemId, startIds[262]);
+      assert.lengthOf(bounded.projection.visibleTurnItems, 75);
+      assert.strictEqual(bounded.projection.visibleTurnItems[0]?.sourceItemId, startIds[393]);
       const loaded = bounded.projection.visibleTurnItems.map((row) => String(row.sourceItemId));
       let cursor = bounded.historyCursor;
       for (let pages = 0; cursor !== null; pages += 1) {
-        assert.isBelow(pages, 5);
+        assert.isBelow(pages, 25);
         const anchor = decodeThreadHistoryCursor(cursor);
         const snapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, {
           rowLimit: 77,
@@ -1441,6 +1444,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
           cursor,
           snapshotSequence: 0,
         });
+        assert.isAtMost(page.items.length, THREAD_HISTORY_PAGE_POLICY.maxItems);
         loaded.unshift(...page.items.map((row) => String(row.sourceItemId)));
         cursor = page.nextCursor;
       }
@@ -2119,7 +2123,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
         snapshotSequence: cancelledSuffixSnapshot.snapshotSequence,
       });
       assert.strictEqual(cancelledSuffixPage.latestLocalTurnOrdinal, 1_202);
-      assert.lengthOf(cancelledSuffixPage.projection.visibleTurnItems, 75);
+      assert.lengthOf(cancelledSuffixPage.projection.visibleTurnItems, 1);
       assert.isFalse(
         cancelledSuffixPage.projection.visibleTurnItems.some((row) =>
           String(row.sourceItemId).includes(":cancelled:"),
@@ -2130,6 +2134,28 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
           (row) => row.sourceItemId === runlessQueuedId,
         ),
       );
+      assert.isTrue(cancelledSuffixPage.hasMoreHistory);
+      assert.isNotNull(cancelledSuffixPage.historyCursor);
+      const queuedSuffixIds = cancelledSuffixPage.projection.visibleTurnItems.map((row) =>
+        String(row.sourceItemId),
+      );
+      let queuedCursor: string | null = cancelledSuffixPage.historyCursor;
+      for (let pages = 0; queuedCursor !== null; pages += 1) {
+        assert.isBelow(pages, 14);
+        const page: OrchestrationV2ThreadHistoryPage = yield* projectionStore.getThreadHistoryPage(
+          threadId,
+          queuedCursor,
+        );
+        queuedSuffixIds.unshift(...page.items.map((row) => String(row.sourceItemId)));
+        queuedCursor = page.nextCursor;
+      }
+      assert.deepEqual(queuedSuffixIds, [
+        ...Array.from(
+          { length: 1_000 },
+          (_, index) => `turn-item:bounded-sql-history:${index + 1}`,
+        ),
+        runlessQueuedId,
+      ]);
 
       const interruptRunId = RunId.make("run:bounded-sql-history:interrupt");
       const interruptNodeId = NodeId.make("node:bounded-sql-history:interrupt");
@@ -5566,6 +5592,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
         projection: yield* projectionStore
           .getThreadSnapshotWindow(nestedThreadId, {
             rowLimit: THREAD_HISTORY_PAGE_POLICY.maxItems + 2,
+            userTurnLimit: 10,
           })
           .pipe(Effect.map((snapshot) => snapshot.projection)),
         snapshotSequence: 1,
@@ -5582,6 +5609,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
         const decoded = decodeThreadHistoryCursor(nestedCursor);
         const snapshot = yield* projectionStore.getThreadSnapshotWindow(nestedThreadId, {
           rowLimit: THREAD_HISTORY_PAGE_POLICY.maxItems + 2,
+          userTurnLimit: 20,
           anchorItemId: TurnItemId.make(decoded.si),
           anchorThreadId: ThreadId.make(decoded.st),
         });

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

@@ -88,9 +88,7 @@ import {
   threadHistoryTurnStarts,
   THREAD_HISTORY_MAX_RAW_TURNS,
   selectHistoryPageFromCursor,
-  selectRecentTimelineWindow,
-  THREAD_HISTORY_PAGE_POLICY,
-  OLDER_THREAD_USER_TURN_LIMIT,
+  projectedRowEncodedBytes,
 } from "./threadHistoryPaging.ts";
 
 export class ProjectionStoreApplyEventError extends Schema.TaggedError<ProjectionStoreApplyEventError>()(
@@ -5073,31 +5071,24 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
           FROM orchestration_events WHERE application_event_version = 2
           AND aggregate_kind = 'thread' AND stream_id = ${threadId}`;
             const snapshotSequence = revisions[0]?.sequence ?? 0;
+            const indexedItems = index.visible.map((row, position) => ({
+              ...row,
+              position,
+              canonicalPosition: position,
+            }));
             const page = yield* Effect.try({
               try: () =>
                 selectHistoryPageFromCursor({
-                  items: index.visible.map((row, position) => ({
-                    ...row,
-                    position,
-                    canonicalPosition: position,
-                  })),
+                  items: indexedItems,
                   cursor,
                   snapshotSequence,
                   rowEncodedBytes: () => 0,
                   throughEntryId,
                 }),
               catch: (cause) => new ProjectionStoreReadError({ threadId, cause }),
             });
-            const end = (page.items.at(-1)?.canonicalPosition ?? -1) + 1;
-            const hasUserTurns = index.visible
-              .slice(0, end)
-              .some((row) => isThreadHistoryUserTurn(row.item));
-            const materialized =
-              conversationOnly && hasUserTurns
-                ? page.items.filter((row) => isConversationHistoryItem(row.item))
-                : page.items;
             const sources = yield* encodeTimelineSources(
-              materialized
+              page.items
                 .filter((row) => !row.synthetic)
                 .map((row) => ({ threadId: row.sourceThreadId, id: row.sourceItemId })),
             );
@@ -5113,7 +5104,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
             const byIdentity = new Map(
               payloads.map((row) => [`${row.thread_id}:${row.turn_item_id}`, row.payload_json]),
             );
-            const items = yield* Effect.forEach(materialized, (row) =>
+            const items = yield* Effect.forEach(page.items, (row) =>
               Effect.gen(function* () {
                 const payload = byIdentity.get(`${row.sourceThreadId}:${row.sourceItemId}`);
                 const item =
@@ -5130,29 +5121,32 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                 };
               }),
             );
-            if (hasUserTurns)
-              return {
-                ...page,
-                snapshotSequence,
-                items: items.map((row, position) => ({ ...row, position })),
-              };
-            const bounded = selectRecentTimelineWindow({
-              items,
+            const wireRowsByPosition = new Map(
+              items.map((row) => [
+                row.position,
+                { row, encodedBytes: projectedRowEncodedBytes(row) },
+              ]),
+            );
+            const bounded = selectHistoryPageFromCursor({
+              items: indexedItems,
+              cursor,
+              throughEntryId,
               snapshotSequence,
-              policy: {
-                ...THREAD_HISTORY_PAGE_POLICY,
-                maxUserTurns: OLDER_THREAD_USER_TURN_LIMIT,
-                maxItems:
-                  THREAD_HISTORY_PAGE_POLICY.maxItems * (throughEntryId === undefined ? 1 : 2),
-              },
+              rowEncodedBytes: (row) =>
+                wireRowsByPosition.get(row.canonicalPosition)?.encodedBytes ?? 0,
             });
+            // Non-negative wire costs can only trim the suffix materialized by the zero-cost pass.
+            const selected = bounded.items.map((row) => ({
+              ...wireRowsByPosition.get(row.canonicalPosition)!.row,
+              position: row.position,
+            }));
             return {
               snapshotSequence,
               items: conversationOnly
-                ? bounded.items.filter((row) => isConversationHistoryItem(row.item))
-                : bounded.items,
-              nextCursor: bounded.hasMoreHistory ? bounded.nextCursor : page.nextCursor,
-              hasMoreHistory: bounded.hasMoreHistory || page.hasMoreHistory,
+                ? selected.filter((row) => isConversationHistoryItem(row.item))
+                : selected,
+              nextCursor: bounded.nextCursor,
+              hasMoreHistory: bounded.hasMoreHistory,
             };
           }),
         )

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

@@ -986,6 +986,84 @@ describe("V2 history materialization", () => {
     }).pipe(Effect.provide(layerTest)),
   );
 
+  it.effect.each(["wake", "steering"] as const)(
+    "preserves full-history turn boundaries when paging %s notifications in SQL",
+    (kind) =>
+      Effect.gen(function* () {
+        const projection = yield* setup;
+        const notificationRun = kind === "wake" ? RunId.make("run:wake") : runId;
+        if (kind === "wake") yield* commit([run(threadId, notificationRun, 2)]);
+        const notification = {
+          ...item("notification", kind === "wake" ? 80 : 82, ""),
+          runId: notificationRun,
+          type: "notification" as const,
+          source: { kind: "monitor" as const },
+          outcome: "completed" as const,
+          summary: kind,
+        };
+        const newestTurn =
+          kind === "wake"
+            ? [
+                notification,
+                ...Array.from({ length: 200 }, (_, index) => ({
+                  ...item(`wake-tool:${index}`, 81 + index, ""),
+                  runId: notificationRun,
+                  type: "command_execution" as const,
+                  input: "pwd",
+                  output: "/work",
+                  exitCode: 0,
+                })),
+              ]
+            : [
+                {
+                  ...item("queued", 80, "next turn", "user_message"),
+                  runId: null,
+                  inputIntent: "queued_turn" as const,
+                },
+                item("answer:before-steer", 81, "a".repeat(700_000)),
+                notification,
+                item("answer:after-steer", 83, "b".repeat(700_000)),
+              ];
+        yield* putItems([
+          item("user", 0, "initial turn", "user_message"),
+          ...Array.from({ length: 79 }, (_, index) => item(`older:${index}`, index + 1, "work")),
+          ...newestTurn,
+        ]);
+        const snapshot = yield* projection.getThreadSnapshot(threadId);
+        const rows = snapshot.projection.visibleTurnItems.map((row) => ({
+          ...row,
+          item: projectTurnItemForWire(row.item),
+        }));
+        const cursor = encodeThreadHistoryCursor({
+          snapshotSequence: snapshot.snapshotSequence,
+          sourceThreadId: threadId,
+          sourceItemId: "absent",
+          position: rows.length,
+        });
+        const expected = selectHistoryPageFromCursor({
+          items: rows,
+          cursor,
+          snapshotSequence: snapshot.snapshotSequence,
+        });
+        const expectedCount = kind === "wake" ? 201 : 4;
+        const expectedFirstId = kind === "wake" ? "notification" : "queued";
+        assert.equal(expected.items.length, expectedCount);
+        assert.equal(expected.items[0]?.sourceItemId, expectedFirstId);
+        assert.deepEqual(
+          expected.items.map((row) => row.sourceItemId),
+          newestTurn.map((entry) => entry.id),
+        );
+        assert.isTrue(expected.hasMoreHistory);
+
+        const page = yield* projection.getThreadHistoryPage(threadId, cursor);
+        assert.deepEqual(
+          { count: page.items.length, firstId: page.items[0]?.sourceItemId },
+          { count: expectedCount, firstId: expectedFirstId },
+        );
+        assert.deepEqual(page, { snapshotSequence: snapshot.snapshotSequence, ...expected });
+      }).pipe(Effect.provide(layerTest)),
+  );
+
   it.effect("enforces fallback item and byte limits and always admits an oversized row", () =>
     Effect.gen(function* () {
       const projection = yield* setup;