fix(agent-runtime,thread-view): stop timeline 500s from session-scoped ACP fs-write ids

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

@@ -31,9 +31,9 @@ import {
   listStoredClientTurnRequestIdsInRange,
   listStoredEventRowsByParentToolCallIds,
   isTimelineCursorSequencePresent,
-  listItemEventSpansByItemIds,
-  listStoredBufferedTextDeltaRowsByItemIds,
-  listStoredItemLifecycleRowsByItemIds,
+  listItemEventSpansByItems,
+  listStoredBufferedTextDeltaRowsByItems,
+  listStoredItemLifecycleRowsByItems,
   listLatestBackgroundTaskStateRowsByItemIds,
   listLatestGoalEventRowsByThreadIds,
   listLatestOpenBackgroundTaskStateRowsForThread,
@@ -44,10 +44,12 @@ import {
   listStoredTurnInputAcceptedRowsByClientRequestIds,
   listStoredTurnStartedRowsByTurnIdsUpToSequence,
   listTimelineSegmentAnchorsDescending,
+  scopedItemRefKey,
 } from "@bb/db";
 import type {
   DbConnection,
   InlineOutputCharLimit,
+  ScopedItemRef,
   StandardTimelineSegmentAnchorRow,
   StoredEventRow,
 } from "@bb/db";
@@ -740,6 +742,18 @@ function ensureSequenceWindowTurnCompletedRows(
     : mergeStoredEventRowsById([...args.rows, ...completedRows]);
 }
 
+/**
+ * The scoped identity of the item a row belongs to. Callers must have checked
+ * that the row carries an item id.
+ */
+function storedEventRowItemRef(row: StoredEventRow): ScopedItemRef {
+  return {
+    itemId: row.itemId ?? "",
+    scopeKind: row.scopeKind,
+    turnId: row.turnId,
+  };
+}
+
 interface SequenceWindowItemRowsArgs extends TimelineWindowRowsArgs {
   /** Exclusive upper bound of the window, or undefined for the latest page. */
   beforeSequence: number | undefined;
@@ -769,47 +783,60 @@ function ensureSequenceWindowWholeItemRows(
   db: DbConnection,
   args: SequenceWindowItemRowsArgs,
 ): StoredEventRow[] {
-  const windowItemIds = new Set<string>();
+  // Keyed by scoped identity, not by item id: providers reuse item ids across
+  // turns (a resumed ACP session restarts its synthetic id counter), and a
+  // thread-wide span for such an id makes every window disown the item.
+  const windowItems = new Map<string, ScopedItemRef>();
   for (const row of args.rows) {
     if (
       row.itemId !== null &&
       row.itemKind !== "backgroundTask" &&
       row.sequence >= args.sequenceStart
     ) {
-      windowItemIds.add(row.itemId);
+      const ref = storedEventRowItemRef(row);
+      windowItems.set(scopedItemRefKey(ref), ref);
     }
   }
-  if (windowItemIds.size === 0) {
+  if (windowItems.size === 0) {
     return [...args.rows];
   }
 
   // Spans, not lifecycle rows. An item emits between its start and its end —
   // output deltas, reasoning text, tool progress — and an unfinished item has
   // no end at all, so "does this item reach past the cut" cannot be answered
   // from `item/started` and `item/completed`.
-  const spans = listItemEventSpansByItemIds(db, {
-    itemIds: [...windowItemIds],
+  const spans = listItemEventSpansByItems(db, {
+    items: [...windowItems.values()],
     threadId: args.threadId,
   });
-  const itemIdsOwnedByNewerWindow = new Set<string>();
-  const itemIdsStartingBeforeWindow = new Set<string>();
+  const itemKeysOwnedByNewerWindow = new Set<string>();
+  const itemsStartingBeforeWindow = new Map<string, ScopedItemRef>();
   for (const span of spans) {
+    const key = scopedItemRefKey(span);
     if (
       args.beforeSequence !== undefined &&
       span.maxSequence >= args.beforeSequence
     ) {
-      itemIdsOwnedByNewerWindow.add(span.itemId);
+      itemKeysOwnedByNewerWindow.add(key);
       continue;
     }
     if (span.minSequence < args.sequenceStart) {
-      itemIdsStartingBeforeWindow.add(span.itemId);
+      itemsStartingBeforeWindow.set(key, {
+        itemId: span.itemId,
+        scopeKind: span.scopeKind,
+        turnId: span.turnId,
+      });
     }
   }
 
   const rows = args.rows.filter(
-    (row) => row.itemId === null || !itemIdsOwnedByNewerWindow.has(row.itemId),
+    (row) =>
+      row.itemId === null ||
+      !itemKeysOwnedByNewerWindow.has(
+        scopedItemRefKey(storedEventRowItemRef(row)),
+      ),
   );
-  if (itemIdsStartingBeforeWindow.size === 0) {
+  if (itemsStartingBeforeWindow.size === 0) {
     return rows;
   }
 
@@ -824,37 +851,40 @@ function ensureSequenceWindowWholeItemRows(
   // snapshot of the message, so dropping the prefix would make text disappear
   // as the event-budget floor advances. Carry that one item's prefix into the
   // owning page until item/completed supplies the canonical final text.
-  const backfillRows = listStoredItemLifecycleRowsByItemIds(db, {
-    itemIds: [...itemIdsStartingBeforeWindow],
+  const backfillRows = listStoredItemLifecycleRowsByItems(db, {
+    items: [...itemsStartingBeforeWindow.values()],
     maxInlineOutputChars: args.maxInlineOutputChars,
     threadId: args.threadId,
   }).filter((row) => row.sequence < args.sequenceStart);
 
-  const completedItemIds = new Set<string>();
+  const completedItemKeys = new Set<string>();
   for (const row of [...rows, ...backfillRows]) {
     if (row.type === "item/completed" && row.itemId !== null) {
-      completedItemIds.add(row.itemId);
+      completedItemKeys.add(scopedItemRefKey(storedEventRowItemRef(row)));
     }
   }
   // Delta rows are stored with a null itemKind, and an item that started below
   // the cut has only delta rows inside the window — so the kind must be read
   // from the backfilled item/started row, never from the in-window rows.
-  const bufferedTextItemIds = new Set<string>();
+  const bufferedTextItems = new Map<string, ScopedItemRef>();
   for (const row of backfillRows) {
+    if (row.type !== "item/started" || row.itemId === null) {
+      continue;
+    }
+    const ref = storedEventRowItemRef(row);
+    const key = scopedItemRefKey(ref);
     if (
-      row.type === "item/started" &&
-      row.itemId !== null &&
-      !completedItemIds.has(row.itemId) &&
+      !completedItemKeys.has(key) &&
       (row.itemKind === "agentMessage" ||
         row.itemKind === "plan" ||
         row.itemKind === "reasoning")
     ) {
-      bufferedTextItemIds.add(row.itemId);
+      bufferedTextItems.set(key, ref);
     }
   }
-  const bufferedTextRows = listStoredBufferedTextDeltaRowsByItemIds(db, {
+  const bufferedTextRows = listStoredBufferedTextDeltaRowsByItems(db, {
     beforeSequence: args.sequenceStart,
-    itemIds: [...bufferedTextItemIds],
+    items: [...bufferedTextItems.values()],
     threadId: args.threadId,
   });
   const prefixRows = [...backfillRows, ...bufferedTextRows];

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

@@ -137,6 +137,151 @@ function insertTurns(
   insertEvents(db, noopNotifier, events);
 }
 
+/**
+ * Builds `turnCount` turns that each end with a file change carrying the *same*
+ * item id. Providers really do this: a resumed ACP session restarts its
+ * synthetic `acp-fs-write-N` counter, so an id from an early turn comes back in
+ * a later one.
+ */
+function insertTurnsWithReusedFileChangeItemId(
+  db: DbConnection,
+  thread: Thread,
+  turnCount: number,
+  fillerItemsPerTurn: number,
+): void {
+  const reusedItemId = "acp-fs-write-1";
+  const events: Parameters<typeof insertEvents>[2] = [];
+  let sequence = 0;
+  const push = (
+    event: Omit<Parameters<typeof insertEvents>[2][number], "sequence">,
+  ): void => {
+    sequence += 1;
+    events.push({ ...event, sequence });
+  };
+
+  for (let turn = 1; turn <= turnCount; turn += 1) {
+    const turnId = `turn-${turn}`;
+    const clientRequestId = requestId(turn);
+    push({
+      threadId: thread.id,
+      type: "client/turn/requested",
+      scope: threadScope(),
+      itemId: null,
+      itemKind: null,
+      data: JSON.stringify({
+        direction: "outbound",
+        source: "tell",
+        initiator: "user",
+        request: { method: "turn/start", params: {} },
+        requestId: clientRequestId,
+        senderThreadId: null,
+        input: [{ type: "text", text: `User message ${turn}`, mentions: [] }],
+        target: turn === 1 ? { kind: "thread-start" } : { kind: "new-turn" },
+        execution,
+      }),
+    });
+    push({
+      threadId: thread.id,
+      type: "turn/started",
+      scope: turnScope(turnId),
+      providerThreadId,
+      itemId: null,
+      itemKind: null,
+      data: JSON.stringify({}),
+    });
+    push({
+      threadId: thread.id,
+      type: "turn/input/accepted",
+      scope: turnScope(turnId),
+      providerThreadId,
+      itemId: null,
+      itemKind: null,
+      data: JSON.stringify({ clientRequestId }),
+    });
+    for (let item = 0; item < fillerItemsPerTurn; item += 1) {
+      push({
+        threadId: thread.id,
+        type: "item/completed",
+        scope: turnScope(turnId),
+        providerThreadId,
+        itemId: `${turnId}-item-${item}`,
+        itemKind: "agentMessage",
+        data: JSON.stringify({
+          item: {
+            type: "agentMessage",
+            id: `${turnId}-item-${item}`,
+            text: `Turn ${turn} item ${item}`,
+          },
+        }),
+      });
+    }
+    const changes = [
+      {
+        path: "src/a.ts",
+        kind: "update",
+        diff: `@@ -1 +1 @@\n-old\n+${turnId}`,
+      },
+    ];
+    for (const type of ["item/started", "item/completed"] as const) {
+      push({
+        threadId: thread.id,
+        type,
+        scope: turnScope(turnId),
+        providerThreadId,
+        itemId: reusedItemId,
+        itemKind: "fileChange",
+        data: JSON.stringify({
+          item: {
+            type: "fileChange",
+            id: reusedItemId,
+            changes,
+            status: type === "item/completed" ? "completed" : "pending",
+            approvalStatus: null,
+          },
+        }),
+      });
+    }
+  }
+  insertEvents(db, noopNotifier, events);
+}
+
+/** Every file-change row the walk can reach, oldest page first. */
+function walkAllFileChangeDiffs(
+  db: DbConnection,
+  thread: Thread,
+  eventBudget: number,
+): string[] {
+  const diffsByPage: string[][] = [];
+  let cursor: TimelinePaginationCursor | null = null;
+  for (let page = 0; page < 200; page += 1) {
+    const response = buildThreadTimeline(db, thread, {
+      eventBudget,
+      includeProviderUnhandledOperations: false,
+      includeNestedRows: true,
+      maxInlineOutputChars: null,
+      maxSeq: 0,
+      page: cursor
+        ? { kind: "older", beforeCursor: cursor, segmentLimit: 20 }
+        : { kind: "latest", segmentLimit: 20 },
+    });
+    diffsByPage.push(
+      response.rows
+        .filter((row) => row.kind === "work" && row.workKind === "file-change")
+        .map((row) =>
+          row.kind === "work" && row.workKind === "file-change"
+            ? (row.change.diff ?? "")
+            : "",
+        ),
+    );
+    if (!response.timelinePage.hasOlderRows) {
+      break;
+    }
+    cursor = response.timelinePage.olderCursor;
+    expect(cursor).not.toBeNull();
+  }
+  return diffsByPage.reverse().flat();
+}
+
 interface WalkResult {
   pages: number;
   userMessages: string[];
@@ -263,6 +408,24 @@ describe("timeline event budget", () => {
     );
   });
 
+  it("keeps one file change per turn when turns reuse a file-change item id", () => {
+    const { db, thread } = setup();
+    // 3 turns of 13 events against a budget of 10, so the cut lands inside a
+    // turn and whole-item closure runs. Read as one thread-wide item, the
+    // reused id spans every turn: the newest page backfills the oldest turn's
+    // lifecycle rows, and every older page disowns the item, so the earlier
+    // file changes vanish from the timeline entirely.
+    insertTurnsWithReusedFileChangeItemId(db, thread, 3, 8);
+
+    const unbudgeted = walkAllFileChangeDiffs(db, thread, LARGE_BUDGET);
+    expect(unbudgeted).toEqual([
+      "@@ -1 +1 @@\n-old\n+turn-1",
+      "@@ -1 +1 @@\n-old\n+turn-2",
+      "@@ -1 +1 @@\n-old\n+turn-3",
+    ]);
+    expect(walkAllFileChangeDiffs(db, thread, 10)).toEqual(unbudgeted);
+  });
+
   it("leaves a thread that fits inside the budget byte-identical", () => {
     const { db, thread } = setup();
     insertTurns(db, thread, 4, 5);