From 2512ebade8b50ad5a2985d679b3968d8c73aad0a Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 16 Oct 2026 12:14:16 -0400 Subject: [PATCH 1/3] fix(messages): keep read results older than a channel's loaded history out of its timeline Agents, search and previews read channel messages through the session, and the store merged every result into the open channel window. An old message from such a read landed above the loaded history with a gap below it. Each older page then filled that gap in the middle of the list, so the timeline neither anchored the reader nor told the rows from new arrivals: the reader jumped back and the Jump to latest button counted the page as new messages. A window now shows rows only from its loaded history and newer. Rows from before it wait for the page that reaches them. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- src/features/relay/store.test.ts | 67 ++++++++++++++++++++++++++++++++ src/features/relay/store.ts | 20 ++++++++-- 2 files changed, 83 insertions(+), 4 deletions(-) diff --git a/src/features/relay/store.test.ts b/src/features/relay/store.test.ts index 36d6e5ade7..f21a12d3d6 100644 --- a/src/features/relay/store.test.ts +++ b/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) => { + 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(); diff --git a/src/features/relay/store.ts b/src/features/relay/store.ts index ca6995aed4..78dbf85cfd 100644 --- a/src/features/relay/store.ts +++ b/src/features/relay/store.ts @@ -283,11 +283,23 @@ export function createChannelStore( function setWindow(state: WindowState, patch: Partial) { 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.cursor?.createdAt + : 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() ?? []) From 01701cdaef1785546ca6ef18dbcfcfbe111dedb1 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 16 Oct 2026 12:29:35 -0400 Subject: [PATCH 2/3] test(agents): add effort to the Goose launch-boundary fixture Main fails typecheck since #1250 made effort required and #1185 added this fixture without it. Same one-line fix as in #1257 and #1260. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- src/features/agents/launch-boundary.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/src/features/agents/launch-boundary.test.ts b/src/features/agents/launch-boundary.test.ts index 8f4f43a186..bbfddddcb8 100644 --- a/src/features/agents/launch-boundary.test.ts +++ b/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, From b082c625175279d9fd6ef44a6657da59e81a0ee7 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 16 Oct 2026 14:01:22 -0400 Subject: [PATCH 3/3] fix(messages): keep loaded rows visible when a reconnect head skips past them The history floor read the window cursor, but a reconnect catch-up moves that cursor to the new head. When more than one page arrived during the outage, every row the reader had loaded fell below the floor and vanished. The window now keeps its own floor: the oldest point a page read reached. It only falls while the window lives and resets when the window's rows are cleared. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- src/features/relay/live-session.test.ts | 68 +++++++++++++++++++++++++ src/features/relay/store.ts | 26 ++++++---- 2 files changed, 85 insertions(+), 9 deletions(-) diff --git a/src/features/relay/live-session.test.ts b/src/features/relay/live-session.test.ts index 73a8f65b38..b7ad35c803 100644 --- a/src/features/relay/live-session.test.ts +++ b/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(); + } +}); diff --git a/src/features/relay/store.ts b/src/features/relay/store.ts index 78dbf85cfd..bc79c06bfe 100644 --- a/src/features/relay/store.ts +++ b/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,6 +284,10 @@ 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) { let rows = patch.rows ?? state.snapshot.rows; if (authorized(state.channelId) && transport) { @@ -288,9 +296,7 @@ export function createChannelStore( // 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.cursor?.createdAt - : undefined; + (patch.hasMore ?? state.snapshot.hasMore) ? state.floor : undefined; const combined = new Map( [...state.events, ...state.traffic.values()] .filter( @@ -378,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)); } @@ -513,6 +520,7 @@ export function createChannelStore( state.generation++; state.controller?.abort(); state.events = []; + state.floor = undefined; state.traffic.clear(); setWindow(state, { status: "error", @@ -644,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; @@ -676,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", @@ -834,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; @@ -2102,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; @@ -2142,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; @@ -2572,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",