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, 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.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..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,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) { 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",