fix(messages): stop the channel timeline jumping back when agents or search read old messages

src/features/agents/launch-boundary.test.ts

@@ -531,6 +531,7 @@ it("starts a Codex thread and a turn from the synced copies; offline, only its c
 const GOOSE_LOCAL = {
   model: "local-model",
   provider: "local-provider",
+  effort: "",
   instructions: "LOCAL INSTRUCTIONS",
   workspace: "/this/device",
   scope: "thread" as const,

src/features/relay/live-session.test.ts

@@ -341,3 +341,71 @@ it.each([1, 2, 130])(
     }
   },
 );
+
+it("a reconnect catch-up head newer than every loaded row keeps those rows on screen", async () => {
+  // The catch-up moves the read cursor to the new head. Rows the reader had
+  // already loaded sit below that cursor and must stay visible.
+  const relay = keypair(),
+    viewer = keypair(),
+    alice = keypair();
+  const wire = scriptedTransport(viewer.pubkey, relay.pubkey);
+  let live!: LiveCallbacks;
+  const owner = createRelaySession(
+    {
+      ...wire.transport,
+      subscribe(callbacks) {
+        live = callbacks;
+        return { update() {}, retry() {}, dispose() {} };
+      },
+    },
+    { prepared: true },
+  );
+  const rows = () =>
+    owner.session.channels.window("c").rows.map((row) => row.id);
+  try {
+    owner.session.channels.ensureList();
+    wire
+      .next()
+      .respond([
+        roster(relay, "c", [viewer.pubkey]),
+        metadata(relay, "c", "C"),
+      ]);
+    await flush();
+    const seen = message(alice, "c", "seen", 10),
+      missed = message(alice, "c", "missed", 30),
+      old = message(alice, "c", "old", 5);
+    owner.session.channels.ensure("c");
+    await flush();
+    wire.next().respond([
+      seen,
+      bounds(relay, "c", "head", {
+        has_more: true,
+        next_cursor: { created_at: 10, id: seen.id },
+      }),
+    ]);
+    await flush();
+    expect(rows()).toEqual([seen.id]);
+    live.state({ status: "connected", routes: [] });
+    live.established("c");
+    await flush();
+    expect(wire.pending.length).toBeGreaterThan(0);
+    while (wire.pending.length) {
+      wire.next().respond([
+        missed,
+        bounds(relay, "c", "head", {
+          has_more: true,
+          next_cursor: { created_at: 30, id: missed.id },
+        }),
+      ]);
+      await flush();
+    }
+    expect(rows()).toEqual([seen.id, missed.id]);
+    // Reads older than any loaded page still wait for paging to reach them.
+    const read = owner.session.read([{ ids: [old.id], limit: 1 }]);
+    wire.next().respond([old]);
+    await read;
+    expect(rows()).toEqual([seen.id, missed.id]);
+  } finally {
+    owner.dispose();
+  }
+});

src/features/relay/store.test.ts

@@ -172,6 +172,73 @@ describe("channel store", () => {
     expect(pending.length).toBe(0);
     expect(listener.mock.calls.length).toBeGreaterThanOrEqual(4);
   });
+  it("keeps rows older than a paged window's history bound out until a page reaches them", async () => {
+    // Agents, search and previews read single old messages through the
+    // session. Merging one into an open window above its loaded history makes
+    // a gap that later pages fill in the middle, which the timeline reads as
+    // new arrivals and does not anchor, so the reader jumps back.
+    const { store, queries, next, pending } = setup();
+    const page = (filter: Record<string, unknown>) => {
+      const index = pending.findIndex((entry) =>
+        Object.entries(filter).every(
+          ([key, value]) =>
+            JSON.stringify(entry.filters[0]?.[key as "limit"]) ===
+            JSON.stringify(value),
+        ),
+      );
+      const [entry] = pending.splice(index, 1);
+      assert.exists(entry);
+      return entry;
+    };
+    const newer = message(alice, "c", "newer", 20),
+      middle = message(alice, "c", "middle", 10),
+      old = message(alice, "c", "old", 5);
+    // Read before the window opens: it also seeds the live tail.
+    const early = store.session.read([{ ids: [old.id], limit: 1 }]);
+    next().respond([old]);
+    await early;
+    queries.ensure("c");
+    next().respond([
+      newer,
+      bounds(relay, "c", "head", {
+        has_more: true,
+        next_cursor: { created_at: 20, id: newer.id },
+      }),
+    ]);
+    await flush();
+    expect(queries.window("c").rows.map((row) => row.id)).toEqual([newer.id]);
+    // Read while the window is open.
+    const late = store.session.read([{ ids: [old.id], limit: 1 }]);
+    page({ ids: [old.id] }).respond([old]);
+    await late;
+    expect(queries.window("c").rows.map((row) => row.id)).toEqual([newer.id]);
+    // A read inside the loaded history still lands.
+    const live = message(alice, "c", "live", 30);
+    const current = store.session.read([{ ids: [live.id], limit: 1 }]);
+    page({ ids: [live.id] }).respond([live]);
+    await current;
+    expect(queries.window("c").rows.map((row) => row.id)).toEqual([
+      newer.id,
+      live.id,
+    ]);
+    queries.loadOlder("c");
+    page({ until: 20 }).respond([
+      middle,
+      old,
+      bounds(relay, "c", `20:${newer.id}`, {
+        has_more: false,
+        next_cursor: null,
+      }),
+    ]);
+    await flush();
+    expect(queries.window("c").rows.map((row) => row.id)).toEqual([
+      old.id,
+      middle.id,
+      newer.id,
+      live.id,
+    ]);
+    store.dispose();
+  });
   it("evicts the least recently ensured window beyond capacity, aborting its request and rejecting late results", async () => {
     const { queries, store, next, pending } = setup(2);
     const evicted = vi.fn();

src/features/relay/store.ts

@@ -44,6 +44,10 @@ type WindowState = {
   cursor: WindowCursor | null;
   events: readonly RelayEvent[];
   atHead: boolean;
+  /** Oldest point any page read reached, in created_at seconds; 0 once a read
+   * reached the start. Only falls while the window lives: a catch-up head can
+   * move `cursor` past rows the reader already has, and those must stay. */
+  floor?: number | undefined;
   generation: number;
   controller?: AbortController | undefined;
   /** Rows came from IndexedDB and no relay read has replaced them. Freshness
@@ -280,14 +284,28 @@ export function createChannelStore(
     });
     notify(listListeners);
   }
+  function moveCursor(state: WindowState, cursor: WindowCursor | null) {
+    state.cursor = cursor;
+    state.floor = Math.min(state.floor ?? Infinity, cursor?.createdAt ?? 0);
+  }
   function setWindow(state: WindowState, patch: Partial<ChannelWindow>) {
     let rows = patch.rows ?? state.snapshot.rows;
     if (authorized(state.channelId) && transport) {
+      // Session reads (agents, search, previews) and the live tail can carry
+      // rows from before the loaded history. Showing one leaves a gap that a
+      // later page fills mid-list, where the timeline can neither anchor the
+      // reader nor tell it from new arrivals. That page shows them instead.
+      const floor =
+        (patch.hasMore ?? state.snapshot.hasMore) ? state.floor : undefined;
       const combined = new Map(
-        [...state.events, ...state.traffic.values()].map((event) => [
-          event.id,
-          event as import("./events").EventData,
-        ]),
+        [...state.events, ...state.traffic.values()]
+          .filter(
+            (event) =>
+              floor === undefined ||
+              event.created_at >= floor ||
+              !channelRowKind(event.kind),
+          )
+          .map((event) => [event.id, event as import("./events").EventData]),
       );
       const ids = new Set(combined.keys());
       for (const item of local?.snapshot() ?? [])
@@ -366,6 +384,7 @@ export function createChannelStore(
     state.generation++;
     state.controller?.abort();
     state.events = [];
+    state.floor = undefined;
     state.snapshot = idleWindow(state.channelId);
     notify(windowListeners.get(state.channelId));
   }
@@ -501,6 +520,7 @@ export function createChannelStore(
         state.generation++;
         state.controller?.abort();
         state.events = [];
+        state.floor = undefined;
         state.traffic.clear();
         setWindow(state, {
           status: "error",
@@ -632,7 +652,7 @@ export function createChannelStore(
       if (prepared && !cursor) {
         const head = await requestHead(state.channelId, "foreground");
         if (!live(state, generation)) return;
-        state.cursor = head.cursor;
+        moveCursor(state, head.cursor);
         state.events = head.events;
         state.atHead = true;
         state.restored = head.restored === true;
@@ -664,7 +684,7 @@ export function createChannelStore(
       }
       state.events = retained;
       state.atHead = !cursor;
-      state.cursor = page.cursor;
+      moveCursor(state, page.cursor);
       state.restored = false;
       setWindow(state, {
         status: "ready",
@@ -822,7 +842,7 @@ export function createChannelStore(
           (state.snapshot.status === "idle" ||
             state.snapshot.status === "loading")
         ) {
-          state.cursor = head.cursor;
+          moveCursor(state, head.cursor);
           state.events = head.events;
           state.atHead = true;
           state.restored = head.restored === true;
@@ -2090,7 +2110,7 @@ export function createChannelStore(
       const state = touch(channelId);
       const retained = prepared ? heads.peek(channelId) : undefined;
       if (retained && ["idle", "error"].includes(state.snapshot.status)) {
-        state.cursor = retained.cursor;
+        moveCursor(state, retained.cursor);
         state.events = retained.events;
         state.atHead = true;
         state.restored = retained.restored === true;
@@ -2130,7 +2150,7 @@ export function createChannelStore(
       }
       const head = prepared ? heads.get(channelId) : undefined;
       if (head) {
-        state.cursor = head.cursor;
+        moveCursor(state, head.cursor);
         state.events = head.events;
         state.atHead = true;
         state.restored = head.restored === true;
@@ -2560,7 +2580,7 @@ export function createChannelStore(
         accept(head.events);
         if (!valid())
           throw new DOMException("Stale live catch-up", "AbortError");
-        if (retained.atHead) retained.cursor = head.cursor;
+        if (retained.atHead) moveCursor(retained, head.cursor);
         setWindow(retained, {
           ...(retained.atHead ? { hasMore: head.hasMore } : {}),
           status: "ready",