perf(desktop): merge archived observer pages a page at a time

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);
+  });
+});

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<unknown> = decryptObserverEvent,
 ): Promise<void> {
   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.

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}`),