diff --git a/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs b/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs index 343ce24133..a97b01583c 100644 --- a/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs +++ b/desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs @@ -1187,3 +1187,125 @@ describe("raw-event-level merge: stateful aggregates across live/archive boundar assert.equal(archived[0].seq, 31); }); }); + +// ── Page-at-a-time archive merge ───────────────────────────────────────────── +// +// The archive window is deliberately uncapped (the MAX_OBSERVER_EVENTS cap +// protects the live per-agent store only), so it grows with the age of the +// channel. Merging a page event-by-event meant a linear dedup scan plus a full +// re-sort per event; these pin the behaviour that must survive doing it once +// per page instead. + +describe("archive page merge", () => { + beforeEach(() => { + resetAgentObserverStore(); + }); + + // Decrypt by raw-event id, so one call can carry a whole page. + function decryptById(bySeq) { + return (raw) => Promise.resolve(bySeq.get(raw.id)); + } + + function page(observerEvents) { + const bySeq = new Map(); + const raw = observerEvents.map((event, index) => { + const id = `${index}`.padStart(64, "e"); + bySeq.set(id, event); + return makeRawEvent({ id }); + }); + return { raw, decrypt: decryptById(bySeq) }; + } + + it("orders a whole page ascending even though the archive returns it newest-first", async () => { + _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); + const newestFirst = [5, 4, 3, 2, 1].map((seq) => + makeObserverEvent({ + seq, + timestamp: `2026-01-01T00:00:0${seq}.000Z`, + }), + ); + const { raw, decrypt } = page(newestFirst); + + await ingestArchivedObserverEvents(raw, decrypt); + + assert.deepEqual( + _testGetArchivedChannelEvents(AGENT_PUBKEY, "chan-1").map((e) => e.seq), + [1, 2, 3, 4, 5], + ); + }); + + it("drops duplicates inside one page and against an already-loaded window", async () => { + _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); + const first = page([ + makeObserverEvent({ seq: 1, timestamp: "2026-01-01T00:00:01.000Z" }), + makeObserverEvent({ seq: 2, timestamp: "2026-01-01T00:00:02.000Z" }), + ]); + await ingestArchivedObserverEvents(first.raw, first.decrypt); + + // seq 2 repeats the loaded window; seq 3 repeats itself within the page. + const second = page([ + makeObserverEvent({ seq: 2, timestamp: "2026-01-01T00:00:02.000Z" }), + makeObserverEvent({ seq: 3, timestamp: "2026-01-01T00:00:03.000Z" }), + makeObserverEvent({ seq: 3, timestamp: "2026-01-01T00:00:03.000Z" }), + ]); + await ingestArchivedObserverEvents(second.raw, second.decrypt); + + assert.deepEqual( + _testGetArchivedChannelEvents(AGENT_PUBKEY, "chan-1").map((e) => e.seq), + [1, 2, 3], + ); + }); + + it("keeps each channel's window separate when one page spans several", async () => { + _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); + const { raw, decrypt } = page([ + makeObserverEvent({ + seq: 2, + timestamp: "2026-01-01T00:00:02.000Z", + channelId: "chan-2", + }), + makeObserverEvent({ seq: 1, timestamp: "2026-01-01T00:00:01.000Z" }), + makeObserverEvent({ + seq: 1, + timestamp: "2026-01-01T00:00:01.000Z", + channelId: "chan-2", + }), + ]); + + await ingestArchivedObserverEvents(raw, decrypt); + + assert.deepEqual( + _testGetArchivedChannelEvents(AGENT_PUBKEY, "chan-1").map((e) => e.seq), + [1], + ); + assert.deepEqual( + _testGetArchivedChannelEvents(AGENT_PUBKEY, "chan-2").map((e) => e.seq), + [1, 2], + ); + }); + + it("merges a page into a large window without dropping or reordering it", async () => { + _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]); + const window = Array.from({ length: 2_000 }, (_, index) => + makeObserverEvent({ + seq: index + 1, + timestamp: new Date(1_760_000_000_000 + index * 1_000).toISOString(), + }), + ); + const loaded = page(window); + await ingestArchivedObserverEvents(loaded.raw, loaded.decrypt); + + const older = page([ + makeObserverEvent({ + seq: 0, + timestamp: new Date(1_760_000_000_000 - 1_000).toISOString(), + }), + ]); + await ingestArchivedObserverEvents(older.raw, older.decrypt); + + const merged = _testGetArchivedChannelEvents(AGENT_PUBKEY, "chan-1"); + assert.equal(merged.length, 2_001); + assert.equal(merged[0].seq, 0, "the older page must sort to the front"); + assert.equal(merged.at(-1).seq, 2_000); + }); +}); diff --git a/desktop/src/features/agents/observerRelayStore.ts b/desktop/src/features/agents/observerRelayStore.ts index 7ae4d0bfc8..96979928b6 100644 --- a/desktop/src/features/agents/observerRelayStore.ts +++ b/desktop/src/features/agents/observerRelayStore.ts @@ -327,41 +327,65 @@ function archiveChannelKey(agentPubkey: string, channelId: string): string { } /** - * Append a decoded archived observer event to the channel-scoped archive - * event journal. Unlike `appendAgentEvent`, this path does NOT cap or trim — - * the channel archive window grows only by explicit paged loads from SQLite, - * so unbounded growth from live relay events is impossible. + * Identity of an archived event for dedup: the `(seq, timestamp)` pair + * `appendAgentEvent` and `mergeObserverEventWindows` also key on. NUL-joined + * so no timestamp spelling can borrow a digit from the seq beside it. + */ +function archiveDedupKey(event: { seq: number; timestamp: string }): string { + return `${event.seq}\u0000${event.timestamp}`; +} + +/** + * Merge a page of decoded archived observer events into the channel-scoped + * archive event journal. Unlike `appendAgentEvent`, this path does NOT cap or + * trim — the channel archive window grows only by explicit paged loads from + * SQLite, so unbounded growth from live relay events is impossible. + * + * Deduplicates on `(seq, timestamp)`, both against the events already in the + * window and within the incoming page. That dedup is *archive-local*: the + * archive window and the live transcript are deliberately separate stores and + * live events never write here, so an archived copy of an event that already + * arrived live is still stored. The two windows are reconciled at read time, + * by `mergeObserverEventWindows`. * - * Deduplicates on `(seq, timestamp)` — identical to `appendAgentEvent` — so - * events that arrive on the live relay before the archive page is loaded are - * silently skipped. The archive window and the live transcript are kept - * strictly separate: live events never write here. + * Takes the whole page at once. Merging event-by-event meant a linear dedup + * scan and a full re-sort of the window per event, so ingesting a page of P + * events into a window of M cost O(P x M) comparisons plus P sorts — and the + * archive window is deliberately uncapped, so M grows with the age of the + * channel. Per page it is now one pass to index the window, one to filter, and + * one sort. * - * Returns `true` if the event was added (state changed), `false` if it was a - * duplicate and was skipped. The caller batches notifications. + * Returns `true` if anything was added (state changed), `false` if every event + * was a duplicate. The caller batches notifications. */ -function appendArchivedChannelEvent( +function appendArchivedChannelEvents( agentPubkey: string, channelId: string, - event: ObserverEvent, + events: ObserverEvent[], ): boolean { + if (events.length === 0) return false; const key = archiveChannelKey(agentPubkey, channelId); const current = archiveEventsByChannel.get(key) ?? []; - // Dedup: skip if (seq, timestamp) already present in the archive window. - if ( - current.some( - (existing) => - existing.seq === event.seq && existing.timestamp === event.timestamp, - ) - ) { - return false; + // One pass over the existing window instead of one scan per incoming event. + // A page merged event-by-event cost O(page x window) in dedup alone, which is + // what made paging back through a long-lived channel slower the further back + // it went. + const seen = new Set(current.map(archiveDedupKey)); + const accepted: ObserverEvent[] = []; + for (const event of events) { + const dedupKey = archiveDedupKey(event); + // Adding as we go also dedups within the incoming page, which the + // per-event version got for free by appending to `current` each time. + if (seen.has(dedupKey)) continue; + seen.add(dedupKey); + accepted.push(event); } + if (accepted.length === 0) return false; - // Archive pages arrive newest-first from SQLite, so each new event sorts - // BEFORE the existing entries. Sort the combined array to maintain ascending - // order for consumers that call buildTranscriptState over the window. - const sorted = [...current, event].sort(compareObserverEvents); + // Archive pages arrive newest-first from SQLite, so the merged array needs a + // sort — but once for the page, not once per event. + const sorted = [...current, ...accepted].sort(compareObserverEvents); archiveEventsByChannel.set(key, sorted); return true; } @@ -775,10 +799,17 @@ export function useManagedAgentObserverBridge( * - The event sender (`pubkey`) must match the `agent` tag value. * - Event must decrypt successfully via `decryptObserverEvent`. * - * Routes through `appendAgentEvent` so dedup on `(seq, timestamp)` and - * sort are reused — archived events that are already present (live-delivered) - * are silently skipped. Failed decryptions are silently dropped (same as - * live path error handling). + * Routing depends on whether the decoded event carries a `channelId`: + * + * - With one, it joins the channel-scoped archive window via + * `appendArchivedChannelEvents`, batched so each channel's slice of the page + * is merged in a single call. Dedup there is archive-local — an event that + * also arrived live is still stored, and the two windows are reconciled at + * read time by `mergeObserverEventWindows`. + * - Without one, it falls through to `appendAgentEvent` so it stays visible in + * the agent's general transcript. + * + * Failed decryptions are silently dropped (same as live path error handling). * * Note: events for agents not currently registered in `knownAgentPubkeys` * (e.g. an agent that is stopped but has archived history) are dropped. @@ -792,6 +823,12 @@ export async function ingestArchivedObserverEvents( _decryptFn: (event: RelayEvent) => Promise = decryptObserverEvent, ): Promise { let archiveChanged = false; + // Collected per (agent, channel) so each channel's slice of the page is + // merged in one call rather than once per event. + const pendingByChannel = new Map< + string, + { agentPubkey: string; channelId: string; events: ObserverEvent[] } + >(); for (const event of rawEvents) { const agentPubkey = observerTag(event, "agent"); const frame = observerTag(event, "frame"); @@ -812,12 +849,17 @@ export async function ingestArchivedObserverEvents( // Events without a channelId fall through to the live store so they // remain visible in the agent's general transcript. if (inner.channelId) { - const added = appendArchivedChannelEvent( - agentPubkey, - inner.channelId, - inner, - ); - if (added) archiveChanged = true; + const channelKey = archiveChannelKey(agentPubkey, inner.channelId); + const pending = pendingByChannel.get(channelKey); + if (pending) { + pending.events.push(inner); + } else { + pendingByChannel.set(channelKey, { + agentPubkey, + channelId: inner.channelId, + events: [inner], + }); + } } else { // Live path already calls notifyListeners() inside appendAgentEvent. appendAgentEvent(agentPubkey, inner); @@ -827,6 +869,11 @@ export async function ingestArchivedObserverEvents( // Silently drop decrypt failures — same as live path error handling. } } + for (const { agentPubkey, channelId, events } of pendingByChannel.values()) { + if (appendArchivedChannelEvents(agentPubkey, channelId, events)) { + archiveChanged = true; + } + } // Batch-notify once for the whole page of archive events. appendAgentEvent // already notifies individually for live/no-channelId events above, so we // only need one extra notify here for the archive path. diff --git a/desktop/src/features/agents/ui/agentSessionPanelLayout.ts b/desktop/src/features/agents/ui/agentSessionPanelLayout.ts index 19d6242842..568d5da144 100644 --- a/desktop/src/features/agents/ui/agentSessionPanelLayout.ts +++ b/desktop/src/features/agents/ui/agentSessionPanelLayout.ts @@ -35,7 +35,7 @@ export function mergeObserverEventWindows( if (archivedEvents.length === 0) return liveEvents as ObserverEvent[]; if (liveEvents.length === 0) return archivedEvents as ObserverEvent[]; - // Dedup key: same as appendAgentEvent / appendArchivedChannelEvent. + // Dedup key: same as appendAgentEvent / appendArchivedChannelEvents. const liveKeySet = new Set(liveEvents.map((e) => `${e.seq}:${e.timestamp}`)); const uniqueArchived = archivedEvents.filter( (e) => !liveKeySet.has(`${e.seq}:${e.timestamp}`),