Key the timeline response cache by request shape, not maxSeq

apps/server/src/routes/threads/data.ts

@@ -56,7 +56,6 @@ import type {
 } from "../../services/threads/timeline-pagination.js";
 import { createSlowThreadTimelineBuildLogger } from "../../services/threads/timeline-build-log.js";
 import {
-  buildThreadTimelineCacheKey,
   buildThreadTimelineParamsKey,
   createThreadTimelineCache,
 } from "../../services/threads/timeline-cache.js";
@@ -355,8 +354,9 @@ export function registerThreadDataRoutes(app: Hono, deps: AppDeps): void {
       summaryOnly,
       includeProviderUnhandledOperations,
     };
+    const paramsKey = buildThreadTimelineParamsKey(keyArgs);
     const full = timelineCache.getOrBuild(
-      buildThreadTimelineCacheKey({ ...keyArgs, maxSeq }),
+      { paramsKey, maxSeq },
       () => {
         const { profile, response } = buildThreadTimelineWithProfile(
           deps.db,
@@ -399,7 +399,6 @@ export function registerThreadDataRoutes(app: Hono, deps: AppDeps): void {
       query.afterSequence,
       "afterSequence",
     );
-    const paramsKey = buildThreadTimelineParamsKey(keyArgs);
     const previous =
       afterSequence === undefined
         ? undefined

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

@@ -13,20 +13,22 @@ import type { ThreadTimelinePageRequest } from "./timeline-pagination.js";
  * (detail view + side-chat tabs), debounced realtime invalidations that fire
  * after the tail already settled, and re-opening a thread.
  *
- * Keying on the thread high-water `maxSeq` makes invalidation implicit: any
- * appended event bumps `maxSeq`, producing a new key and a cold rebuild. The
- * key MUST also include every other input the projection depends on:
+ * Entries are keyed by request shape (`paramsKey`) and store the thread
+ * high-water `maxSeq` they were built at. A request with a different `maxSeq`
+ * is a miss that *replaces* the slot: `maxSeq` never decreases, so the old
+ * revision could never be looked up again and keeping it until global LRU
+ * eviction only pins a dead response per appended event (#2066). The request
+ * shape MUST include every other input the projection depends on:
  * `thread.status` (interrupt flips earlier rows), `environmentId` (workspace
  * root relativizes file paths), provider display name (labels dynamic-provider
  * diagnostic rows), and the row-shape request flags. Event pruning
  * (`pruneResolvedItemDeltas`, background-task progress) is output-preserving
  * and never lowers `maxSeq`, so it cannot stale a cached entry.
  *
  * Entries with many rows are not cached: an expanded active turn (the streaming
- * case) produces hundreds of rows AND a `maxSeq` that changes on every event,
- * so caching it only thrashes the LRU and pins large objects for no reuse. Idle
- * windows collapse completed turns to a handful of rows regardless of thread
- * size, so the cap excludes exactly the entries that would never be reused.
+ * case) produces hundreds of rows that are rebuilt on every event, so storing
+ * them pins a large object for no reuse. The per-shape slot bounds the *count*
+ * of retained revisions; the row cap bounds their *size*.
  */
 
 const DEFAULT_MAX_ENTRIES = 128;
@@ -40,7 +42,7 @@ interface ThreadTimelineCacheOptions {
 
 interface ThreadTimelineCache {
   getOrBuild(
-    key: string,
+    key: { paramsKey: string; maxSeq: number },
     build: () => ThreadTimelineResponse,
   ): ThreadTimelineResponse;
   /** Number of currently cached entries (for tests/metrics). */
@@ -53,21 +55,27 @@ export function createThreadTimelineCache(
   const maxEntries = options.maxEntries ?? DEFAULT_MAX_ENTRIES;
   const maxCacheableRows =
     options.maxCacheableRows ?? DEFAULT_MAX_CACHEABLE_ROWS;
-  const entries = new Map<string, ThreadTimelineResponse>();
+  const entries = new Map<
+    string,
+    { maxSeq: number; value: ThreadTimelineResponse }
+  >();
 
   return {
-    getOrBuild(key, build) {
-      const cached = entries.get(key);
-      if (cached !== undefined) {
+    getOrBuild({ paramsKey, maxSeq }, build) {
+      const cached = entries.get(paramsKey);
+      if (cached?.maxSeq === maxSeq) {
         // Re-insert to mark most-recently-used.
-        entries.delete(key);
-        entries.set(key, cached);
-        return cached;
+        entries.delete(paramsKey);
+        entries.set(paramsKey, cached);
+        return cached.value;
       }
 
       const value = build();
+      // A newer revision supersedes the stored one even when the new value is
+      // too large to cache: the old one can never hit again.
+      entries.delete(paramsKey);
       if (value.rows.length <= maxCacheableRows) {
-        entries.set(key, value);
+        entries.set(paramsKey, { maxSeq, value });
         while (entries.size > maxEntries) {
           const oldest = entries.keys().next().value;
           if (oldest === undefined) {
@@ -86,8 +94,6 @@ export function createThreadTimelineCache(
 
 export interface ThreadTimelineCacheKeyArgs {
   threadId: string;
-  /** Thread high-water event sequence; bumps on every appended event. */
-  maxSeq: number;
   status: ThreadStatus;
   environmentId: string | null;
   providerDisplayName?: string;
@@ -104,12 +110,12 @@ function pageKeyPart(page: ThreadTimelinePageRequest): string {
 }
 
 /**
- * The cache identity *excluding* `maxSeq` — i.e. everything that selects which
- * window is being requested, but not which revision of it. Used to track the
- * latest-sent rows per request shape for delta computation.
+ * The request shape: everything that selects which window is being requested,
+ * but not which revision (`maxSeq`) of it. Shared by the response cache and the
+ * latest-rows delta cache.
  */
 export function buildThreadTimelineParamsKey(
-  args: Omit<ThreadTimelineCacheKeyArgs, "maxSeq">,
+  args: ThreadTimelineCacheKeyArgs,
 ): string {
   return [
     args.threadId,
@@ -122,9 +128,3 @@ export function buildThreadTimelineParamsKey(
     args.includeProviderUnhandledOperations ? "1" : "0",
   ].join("|");
 }
-
-export function buildThreadTimelineCacheKey(
-  args: ThreadTimelineCacheKeyArgs,
-): string {
-  return `${args.maxSeq}|${buildThreadTimelineParamsKey(args)}`;
-}

apps/server/test/public/public-thread-timeline-cache-retention.test.ts

@@ -0,0 +1,130 @@
+/**
+ * Regression for #2066: the timeline response cache used to key entries by
+ * `${maxSeq}|${paramsKey}`, so every appended event stranded the previous
+ * revision of the same request shape in the LRU until global eviction. A
+ * thread's `maxSeq` is monotonic, so those revisions could never be hit again.
+ *
+ * Drives the real `GET /api/v1/threads/:id/timeline` route against in-memory
+ * SQLite. The only instrumentation is wrapping the cache factory so the test
+ * can read `.size` of the instance the route creates.
+ */
+import { describe, expect, it, vi } from "vitest";
+import { threadScope, turnScope } from "@bb/domain";
+import { threadTimelineResponseSchema } from "@bb/server-contract";
+import { readJson } from "../helpers/json.js";
+import { seedEvent, seedThreadFixture } from "../helpers/seed.js";
+import { withTestHarness } from "../helpers/test-app.js";
+import type { TestAppHarness } from "../helpers/test-app.js";
+import type { createThreadTimelineCache as CreateCache } from "../../src/services/threads/timeline-cache.js";
+
+const createdCaches: ReturnType<typeof CreateCache>[] = [];
+
+vi.mock(
+  "../../src/services/threads/timeline-cache.js",
+  async (importOriginal) => {
+    const mod =
+      await importOriginal<
+        typeof import("../../src/services/threads/timeline-cache.js")
+      >();
+    return {
+      ...mod,
+      createThreadTimelineCache: (
+        ...args: Parameters<typeof mod.createThreadTimelineCache>
+      ) => {
+        const cache = mod.createThreadTimelineCache(...args);
+        createdCaches.push(cache);
+        return cache;
+      },
+    };
+  },
+);
+
+async function fetchTimeline(harness: TestAppHarness, threadId: string) {
+  const response = await harness.app.request(
+    `/api/v1/threads/${threadId}/timeline`,
+  );
+  expect(response.status).toBe(200);
+  return threadTimelineResponseSchema.parse(await readJson(response));
+}
+
+describe("GET /threads/:id/timeline response cache retention (#2066)", () => {
+  it("keeps one resident revision per request shape as events are appended", async () => {
+    await withTestHarness(async (harness) => {
+      const cache = createdCaches.at(-1);
+      if (!cache) {
+        throw new Error("route did not create a timeline cache");
+      }
+      const { environment, thread } = seedThreadFixture(harness);
+      const base = {
+        threadId: thread.id,
+        environmentId: environment.id,
+        providerThreadId: "p1",
+      };
+
+      seedEvent(harness.deps, {
+        ...base,
+        scope: threadScope(),
+        sequence: 1,
+        type: "system/manager/user_message",
+        data: { text: "hello" },
+      });
+      seedEvent(harness.deps, {
+        ...base,
+        scope: turnScope("turn-1"),
+        sequence: 2,
+        type: "turn/started",
+        data: {},
+      });
+      seedEvent(harness.deps, {
+        ...base,
+        scope: turnScope("turn-1"),
+        sequence: 3,
+        type: "item/completed",
+        data: { item: { type: "agentMessage", id: "a-1", text: "done" } },
+      });
+      seedEvent(harness.deps, {
+        ...base,
+        scope: turnScope("turn-1"),
+        sequence: 4,
+        type: "turn/completed",
+        data: { status: "completed" },
+      });
+
+      const first = await fetchTimeline(harness, thread.id);
+      expect(first.maxSeq).toBe(4);
+      expect(cache.size).toBe(1);
+
+      // A second, streaming turn: each appended event bumps maxSeq and the
+      // client refetches the same window (same request shape). Enough rounds
+      // to exceed the 128-entry LRU bound if superseded revisions were kept.
+      seedEvent(harness.deps, {
+        ...base,
+        scope: turnScope("turn-2"),
+        sequence: 5,
+        type: "turn/started",
+        data: {},
+      });
+      const rounds = 150;
+      for (let i = 0; i < rounds; i++) {
+        seedEvent(harness.deps, {
+          ...base,
+          scope: turnScope("turn-2"),
+          sequence: 6 + i,
+          type: "item/completed",
+          data: {
+            item: { type: "agentMessage", id: `a-2-${i}`, text: `chunk ${i}` },
+          },
+        });
+        const page = await fetchTimeline(harness, thread.id);
+        expect(page.maxSeq).toBe(6 + i);
+        // Every revision must stay under the row cap so each one is cacheable;
+        // otherwise the assertion below would pass for the wrong reason.
+        expect(page.rows.length).toBeLessThanOrEqual(200);
+      }
+
+      // Only the entry built at the newest maxSeq can ever be hit again.
+      // Before the fix this was 128 (the LRU bound) of dead revisions.
+      expect(cache.size).toBe(1);
+    });
+  });
+});