Compact completed items at their first lifecycle position

apps/server/src/services/threads/stored-event-decode-cache.ts

@@ -1,4 +1,8 @@
-import type { DbConnection, StoredEventRow } from "@bb/db";
+import type {
+  DbConnection,
+  StoredEventRow,
+  ProjectionStoredEventRow,
+} from "@bb/db";
 import type { ThreadEvent } from "@bb/domain";
 import { parseStoredEvent } from "./thread-data.js";
 
@@ -119,7 +123,7 @@ function rememberEntry(
 
 export function decodeStoredEventRowCached(
   db: DbConnection,
-  row: StoredEventRow,
+  row: ProjectionStoredEventRow,
 ): ThreadEvent {
   const cache = getDecodeCache(db);
   const byRow = cache.byRow.get(row);

apps/server/src/services/threads/thread-data.ts

@@ -6,7 +6,11 @@ import {
   hydrateRetainedEventOutputRows,
   listStoredEventRows as listStoredEventRowRecords,
 } from "@bb/db";
-import type { DbConnection, StoredEventRow } from "@bb/db";
+import type {
+  DbConnection,
+  StoredEventRow,
+  ProjectionStoredEventRow,
+} from "@bb/db";
 import { toRecord } from "@bb/core-ui";
 import { buildThreadEventRow, parseStoredThreadEvent } from "@bb/domain";
 import { threadScope, turnScope } from "@bb/domain";
@@ -21,8 +25,8 @@ import { ApiError } from "../../errors.js";
 const THREAD_EVENT_RESPONSE_DATA_BYTE_LIMIT = 8 * 1024 * 1024;
 
 type StoredEventPayloadRow = Pick<
-  StoredEventRow,
-  "data" | "sequence" | "threadId" | "type"
+  ProjectionStoredEventRow,
+  "data" | "sequence" | "threadId" | "type" | "parsedData"
 >;
 
 interface ListThreadEventRowsArgs {
@@ -43,6 +47,7 @@ interface FindThreadEventArgs {
 export function parseStoredEventPayload(
   row: StoredEventPayloadRow,
 ): Record<string, unknown> {
+  if (row.parsedData !== undefined) return row.parsedData;
   let data: unknown;
   try {
     data = JSON.parse(row.data);
@@ -88,7 +93,7 @@ function parseStoredEventScope(row: StoredEventRow): ThreadEventScope {
   }
 }
 
-export function parseStoredEvent(row: StoredEventRow): ThreadEvent {
+export function parseStoredEvent(row: ProjectionStoredEventRow): ThreadEvent {
   return parseStoredThreadEvent({
     type: row.type,
     data: parseStoredEventPayload(row),

apps/server/src/services/threads/thread-fork-history.ts

@@ -1,3 +1,4 @@
+import { expandSelectedCompletedItemRows } from "@bb/db";
 import {
   copyStoredThreadEventsInTransaction,
   findLastCompletedRootStoredTurn,
@@ -262,11 +263,15 @@ function selectInheritedForkEventRows(
   deps: Pick<AppDeps, "db">,
   args: { historyEndSequence: number; sourceThreadId: string },
 ): StoredEventRow[] {
-  const rows = listStoredEventRows(deps.db, {
-    beforeSequence: args.historyEndSequence + 1,
-    threadId: args.sourceThreadId,
-    types: INHERITED_EVENT_TYPES,
-  });
+  const rows = expandSelectedCompletedItemRows(
+    deps.db,
+    listStoredEventRows(deps.db, {
+      beforeSequence: args.historyEndSequence + 1,
+      threadId: args.sourceThreadId,
+      types: INHERITED_EVENT_TYPES,
+    }),
+    args.historyEndSequence,
+  ).filter((row) => INHERITED_EVENT_TYPES.some((type) => type === row.type));
   const completedTurnIds = new Set<string>();
   const acceptedClientRequestIds = new Set<string>();
   for (const row of rows) {

apps/server/src/services/threads/timeline-selection-memo.ts

@@ -136,7 +136,7 @@ function getSelectionMemo(db: DbConnection): TimelineSelectionMemo {
 function dataCharsOfRows(rows: readonly StoredEventRow[]): number {
   let chars = 0;
   for (const row of rows) {
-    chars += row.data.length;
+    chars += row.data.length + (row.completedItemHistory?.length ?? 0);
   }
   return chars;
 }
@@ -280,7 +280,9 @@ export function lookupLatestTimelineSelection(
   const budgetFloor: TimelineBudgetFloor = {
     sequence: findTimelineWindowBudgetFloorSequence(db, {
       threadId: args.threadId,
-      sequenceStart: args.epochSequenceStart,
+      sequenceStart:
+        entry.hints[args.page.segmentLimit]?.sequence ??
+        args.epochSequenceStart,
       beforeSequence: args.maxSeq + 1,
       eventBudget: args.eventBudget,
       excludedTypes: THREAD_TIMELINE_EXCLUDED_EVENT_TYPES,

apps/server/src/services/threads/timeline.ts

@@ -1,3 +1,4 @@
+import { expandSelectedCompletedItemRowsForProjection } from "@bb/db";
 import { paginateTimelineContents } from "./timeline-content-pagination.js";
 import {
   getTimelineGroupingContext,
@@ -972,7 +973,8 @@ function selectStandardTimelineEventRows(
     knownBudgetFloor === null
       ? findTimelineWindowBudgetFloorSequence(db, {
           threadId: thread.id,
-          sequenceStart: epochSequenceStart,
+          sequenceStart:
+            hints[page.segmentLimit]?.sequence ?? epochSequenceStart,
           beforeSequence,
           eventBudget,
           excludedTypes: THREAD_TIMELINE_EXCLUDED_EVENT_TYPES,
@@ -1357,7 +1359,11 @@ function buildThreadTimelineInternal(
           rows: hydrateRetainedEventOutputRows(db, storedEventSelection.rows),
         }
       : storedEventSelection;
-  const rawEventRows = eventSelection.rows;
+  const rawEventRows = expandSelectedCompletedItemRowsForProjection(
+    db,
+    eventSelection.rows,
+    snapshot.maxSeq,
+  );
   profile.eventDataBytes = byteLengthOfStoredEventRows(rawEventRows);
   profile.eventRowCount = rawEventRows.length;
   profile.selectionStrategy = eventSelection.strategy;
@@ -1624,9 +1630,10 @@ export function buildThreadConversationOutline(
       sequenceStart: contextBoundarySeq ?? 0,
       threadId: thread.id,
     });
-    const decodedRawEvents = rawEventRows.map((row) =>
-      toThreadEventWithMeta(row),
-    );
+    const decodedRawEvents = expandSelectedCompletedItemRowsForProjection(
+      db,
+      rawEventRows,
+    ).map((row) => toThreadEventWithMeta(row));
     const decodedEvents = compactThreadTimelineSummaryEvents(decodedRawEvents);
     const clientRequestContextRows = selectClientRequestContextRows(db, {
       rows: rawEventRows,
@@ -1926,7 +1933,11 @@ function buildTimelineTurnSummaryDetailsPage(
         : sourceSeqStart,
     sourceRange.sourceSeqStart,
   );
-  const projectionEvents = projectionEventRows
+  const projectionEvents = expandSelectedCompletedItemRowsForProjection(
+    db,
+    projectionEventRows,
+    snapshot.maxSeq,
+  )
     .filter((row) => row.sequence <= snapshot.maxSeq)
     .map((row) => toThreadEventWithMeta(row));
   const children = buildThreadTimelineTurnDetailsFromEvents({

apps/server/test/services/threads/timeline-event-budget.test.ts

@@ -18,6 +18,8 @@ import {
 } from "@bb/domain";
 import type { ClientTurnRequestId, Thread } from "@bb/domain";
 import {
+  advanceThreadPruning,
+  listStoredEventRows,
   createConnection,
   createProject,
   createThread,
@@ -1390,3 +1392,123 @@ it("resolves acceptance after the next conversation boundary", () => {
   expect(rows).toEqual(expected.rows);
   db.$client.close();
 });
+
+it.each([1, 2, 5, 20])(
+  "traverses compacted items completely with event budget %i and a small byte budget",
+  (eventBudget) => {
+    const { db, thread } = setup();
+    try {
+      insertTurns(db, thread, 3, [2, 15, 2]);
+      db.$client
+        .prepare("UPDATE events SET sequence = -sequence WHERE thread_id = ?")
+        .run(thread.id);
+      db.$client
+        .prepare(
+          "UPDATE events SET sequence = -10 * sequence WHERE thread_id = ?",
+        )
+        .run(thread.id);
+      const owners = listStoredEventRows(db, {
+        threadId: thread.id,
+        types: ["item/completed"],
+      });
+      const additions: Parameters<typeof insertEvents>[2] = [];
+      const ends = new Map<string, number>();
+      for (const row of owners) {
+        if (!row.turnId || !row.itemId)
+          throw new Error("Missing fixture scope");
+        const base = {
+          threadId: thread.id,
+          scope: turnScope(row.turnId),
+          providerThreadId,
+          itemId: row.itemId,
+          itemKind: row.itemKind,
+          parentToolCallId: null,
+        };
+        const data = JSON.parse(row.data);
+        additions.push({
+          ...base,
+          type: "item/started",
+          sequence: row.sequence - 2,
+          createdAt: row.createdAt - 2,
+          data: JSON.stringify({ item: { ...data.item, text: "" } }),
+        });
+        additions.push({
+          ...base,
+          type: "item/agentMessage/delta",
+          itemKind: null,
+          sequence: row.sequence - 1,
+          createdAt: row.createdAt - 1,
+          data: JSON.stringify({ itemId: row.itemId, delta: data.item.text }),
+        });
+        ends.set(row.turnId, row.sequence + 1);
+      }
+      for (const [turnId, sequence] of ends)
+        additions.push({
+          threadId: thread.id,
+          scope: turnScope(turnId),
+          providerThreadId,
+          itemId: null,
+          itemKind: null,
+          parentToolCallId: null,
+          type: "turn/completed",
+          sequence,
+          data: JSON.stringify({ status: "completed" }),
+        });
+      insertEvents(
+        db,
+        noopNotifier,
+        additions.sort((a, b) => a.sequence - b.sequence),
+      );
+      const options = {
+        completedTurnDisplay: "collapse",
+        includeDiagnosticOperations: false,
+        includeNestedRows: true,
+        maxInlineOutputChars: null,
+        maxSeq: 0,
+      } as const;
+      const canonical = buildThreadTimelineWithProfile(db, thread, {
+        ...options,
+        eventBudget: LARGE_BUDGET,
+        page: { kind: "latest", segmentLimit: 100 },
+      }).response.rows;
+      let removed = 0;
+      for (let i = 0; i < 100; i++) {
+        const result = advanceThreadPruning(db, "completed-items");
+        removed += result.removed;
+        if (result.action === "cycle-complete") break;
+      }
+      expect(removed).toBe(owners.length * 2);
+      let rows: TimelineRow[] = [];
+      let cursor: TimelinePaginationCursor | null = null;
+      const cursors = new Set<string>();
+      do {
+        const response: ThreadTimelineResponse = buildThreadTimelineWithProfile(
+          db,
+          thread,
+          {
+            ...options,
+            eventBudget,
+            responseByteBudget: 512,
+            page: cursor
+              ? { kind: "older", beforeCursor: cursor, segmentLimit: 2 }
+              : { kind: "latest", segmentLimit: 2 },
+          },
+        ).response;
+        rows = prependOlderTimelineRows({
+          loadedRows: rows,
+          olderRows: response.rows,
+        });
+        cursor = response.timelinePage.olderCursor;
+        if (cursor) {
+          const key = JSON.stringify(cursor);
+          expect(cursors.has(key)).toBe(false);
+          cursors.add(key);
+          expect(cursors.size).toBeLessThan(100);
+        }
+      } while (cursor);
+      expect(rows).toEqual(canonical);
+    } finally {
+      db.$client.close();
+    }
+  },
+);

apps/server/test/system/event-pruning.test.ts

@@ -1,4 +1,6 @@
-import { getThread, listEvents } from "@bb/db";
+import { getThread, listEvents, threads } from "@bb/db";
+import { eq } from "drizzle-orm";
+import { runThreadPruningSweep } from "../../src/services/system/thread-pruning-sweep.js";
 import { turnScope } from "@bb/domain";
 import { groupHostDaemonEvents } from "@bb/host-daemon-contract";
 import { describe, expect, it, vi } from "vitest";
@@ -171,6 +173,93 @@ function seedResolvedAssistantMessage(
 }
 
 describe("thread event pruning", () => {
+  it("refreshes a cached visible timeline after an idle background completed-item rewrite and exposes combined raw records", async () => {
+    await withTestHarness(async (harness) => {
+      const host = seedHost(harness.deps);
+      const { project } = seedProjectWithSource(harness.deps, {
+        hostId: host.id,
+      });
+      const environment = seedEnvironment(harness.deps, {
+        hostId: host.id,
+        projectId: project.id,
+      });
+      const thread = seedThread(harness.deps, {
+        projectId: project.id,
+        environmentId: environment.id,
+      });
+      seedStoredEvent(harness.deps, {
+        threadId: thread.id,
+        sequence: 1,
+        scope: turnScope("turn-1"),
+        type: "turn/started",
+        itemId: null,
+        itemKind: null,
+        data: {},
+      });
+      seedResolvedAssistantMessage(harness, {
+        threadId: thread.id,
+        itemId: "message",
+        deltaSequences: [2],
+        completedSequence: 3,
+      });
+      seedStoredEvent(harness.deps, {
+        threadId: thread.id,
+        sequence: 4,
+        scope: turnScope("turn-1"),
+        type: "turn/completed",
+        itemId: null,
+        itemKind: null,
+        data: { status: "completed" },
+      });
+      harness.db.$client
+        .prepare("UPDATE events SET provider_thread_id = ? WHERE thread_id = ?")
+        .run("provider", thread.id);
+      const build = () =>
+        buildThreadTimelineWithProfile(harness.db, thread, {
+          completedTurnDisplay: "collapse",
+          includeDiagnosticOperations: false,
+          includeNestedRows: true,
+          maxInlineOutputChars: 8000,
+          eventBudget: 10000,
+          maxSeq: 4,
+          page: { kind: "latest", segmentLimit: 20 },
+        }).response;
+      const before = build();
+      expect(JSON.stringify(before)).toContain("Final answer");
+      expect(build()).toEqual(before);
+      const notify = vi.spyOn(harness.deps.hub, "notifyThread");
+      for (let i = 0; i < 10; i++) {
+        const result = pruneThreadEventHistoryBestEffort(harness.deps, {
+          threadId: thread.id,
+          mode: "active",
+        });
+        expect(result?.policy).not.toBe("completed-items");
+      }
+      expect(listEvents(harness.db, { threadId: thread.id })).toHaveLength(4);
+      expect(notify).not.toHaveBeenCalled();
+      harness.db
+        .update(threads)
+        .set({ status: "idle" })
+        .where(eq(threads.id, thread.id))
+        .run();
+      for (let i = 0; i < 5; i++) await runThreadPruningSweep(harness.deps);
+      expect(notify).toHaveBeenCalledWith(thread.id, ["history-rewritten"]);
+      expect(build()).toEqual(before);
+      const raw = await harness.app.request(
+        `/api/v1/threads/${thread.id}/events?limit=1&afterSeq=1`,
+      );
+      expect(raw.status).toBe(200);
+      const text = await raw.text();
+      expect(text).toContain("item/completed");
+      expect(text).not.toContain("completed_item_history");
+      expect(
+        listEvents(harness.db, { threadId: thread.id }).map(
+          (row) => row.sequence,
+        ),
+      ).toEqual([1, 2, 4]);
+    });
+  });
+
   it("prunes idle-thread noise rows and resolved item deltas", async () => {
     await withTestHarness(async (harness) => {
       const host = seedHost(harness.deps);