fix(desktop): repair observer-store rebase integration with #5680 batching

Rebasing the bounded/tiered observer window onto #5680's per-envelope batch
ingestion left two defects:

- appendAgentEvents kept a per-event recordChannelActivity(key, event) call
  from the pre-#5680 singular appendAgentEvent, but its parameter is now the
  events array — 'event' resolved to the global DOM Event. Record activity for
  every event in the batch so the durable channel-activity summary sees each
  frame.
- #5680's batch dedup/merge/cap logic is more verbose than the singular path it
  replaced, pushing observerRelayStore.ts back over the 1000-line ratchet
  (1032) the channelActivitySummary/observerEventOrdering split existed to
  satisfy. Extract the pure batch computation as mergeObserverEventBatch in
  observerEventOrdering.ts and the latest-live-session subsystem as
  latestLiveSession.ts, both re-exported from the store to keep import paths
  stable. Store lands at 973 lines.

The batching and the tiered per-agent cap compose unchanged: the batch fold
runs against agentEventCap(key), and the incremental-vs-rebuild transcript
decision is preserved in canFoldIncrementally.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>

desktop/src/features/agents/latestLiveSession.ts

@@ -0,0 +1,65 @@
+import { normalizePubkey } from "@/shared/lib/pubkey";
+
+import { isObserverEventAfter } from "./observerEventOrdering";
+import type { ObserverEvent } from "./ui/agentSessionTypes";
+
+// Per-agent, per-channel latest-live-session-id.
+// Key: `${normalizePubkey(agentPubkey)}:${channelId}`.
+// Set when a live relay observer event with a sessionId arrives.
+//
+// "Latest-live" means: the sessionId that most recently appeared via the
+// live relay path (handleRelayObserverEvent). It is NOT derived from
+// connectionState or an ever-live Set — an ever-live Set would incorrectly
+// mark session A as "current" after session B has started (Thufir Pass 3).
+//
+// Stored as `{ sessionId, timestamp, seq }` so that late-arriving live frames
+// from an older session never regress the latest-live id. We only advance when
+// the parsed event sorts strictly AFTER the stored one, using the same
+// two-key ordering as `compareObserverEvents`: timestamp first, then seq on a
+// tie — so a higher-seq frame at equal timestamp still advances the entry.
+type LatestLiveEntry = { sessionId: string; timestamp: string; seq: number };
+const latestLiveSessionByAgentChannel = new Map<string, LatestLiveEntry>();
+
+function liveSessionKey(agentPubkey: string, channelId: string | null): string {
+  return `${normalizePubkey(agentPubkey)}:${channelId ?? ""}`;
+}
+
+/**
+ * Advance the latest-live-session-id for a (agent, channel) pair from one live
+ * event. No-op unless the event carries both a sessionId and channelId (so a
+ * session is never attributed to the wrong channel) and sorts strictly after
+ * the stored entry (so a late frame from an older session cannot regress it).
+ */
+export function recordLatestLiveSession(
+  agentPubkey: string,
+  event: ObserverEvent,
+): void {
+  if (!event.sessionId || !event.channelId) return;
+  const key = liveSessionKey(agentPubkey, event.channelId);
+  const stored = latestLiveSessionByAgentChannel.get(key);
+  if (!stored || isObserverEventAfter(event, stored)) {
+    latestLiveSessionByAgentChannel.set(key, {
+      sessionId: event.sessionId,
+      timestamp: event.timestamp,
+      seq: event.seq,
+    });
+  }
+}
+
+/** Read the latest-live-session-id for a (agent, channel) pair. */
+export function getLatestLiveSessionId(
+  agentPubkey: string | null | undefined,
+  channelId: string | null | undefined,
+): string | null {
+  if (!agentPubkey) return null;
+  return (
+    latestLiveSessionByAgentChannel.get(
+      liveSessionKey(agentPubkey, channelId ?? null),
+    )?.sessionId ?? null
+  );
+}
+
+/** Clear all latest-live-session bookkeeping (store reset). */
+export function clearLatestLiveSessions(): void {
+  latestLiveSessionByAgentChannel.clear();
+}

desktop/src/features/agents/observerEventOrdering.ts

@@ -64,3 +64,57 @@ export function unwrapObserverBatch(parsed: ObserverEvent): ObserverEvent[] {
     : null;
   return events && events.length > 0 ? events : [parsed];
 }
+
+/** Dedup identity for a retained event: length-prefixed timestamp plus seq. */
+function eventDedupKey(event: ObserverEvent): string {
+  return `${event.timestamp.length}:${event.timestamp}:${event.seq}`;
+}
+
+export type ObserverEventBatchMerge = {
+  /** The retained window after dedup, merge, and cap trim. */
+  final: ObserverEvent[];
+  /** The newly added events in ascending order, for an incremental fold. */
+  addedInOrder: ObserverEvent[];
+  /**
+   * True when every added event sorts after the prior tail and no cap trim
+   * fired, so the transcript can fold the batch instead of rebuilding.
+   */
+  canFoldIncrementally: boolean;
+};
+
+/**
+ * Merge an incoming event batch into a retained window: drop events already
+ * present by `(timestamp, seq)`, sort the additions, splice them into `current`
+ * in total order, and trim to `cap` (keeping the most recent). Returns `null`
+ * when every incoming event was a duplicate (no state change). Pure — the store
+ * owns the map writes and transcript state.
+ */
+export function mergeObserverEventBatch(
+  current: readonly ObserverEvent[],
+  incoming: readonly ObserverEvent[],
+  cap: number,
+): ObserverEventBatchMerge | null {
+  const seen = new Set(current.map(eventDedupKey));
+  const added: ObserverEvent[] = [];
+  for (const event of incoming) {
+    const key = eventDedupKey(event);
+    if (seen.has(key)) continue;
+    seen.add(key);
+    added.push(event);
+  }
+  if (added.length === 0) return null;
+
+  const addedInOrder = added.sort(compareObserverEvents);
+  const sorted = [...current, ...addedInOrder].sort(compareObserverEvents);
+  const trimmed = sorted.length > cap;
+  const final = trimmed ? sorted.slice(sorted.length - cap) : sorted;
+
+  const currentLast = current.at(-1);
+  const canFoldIncrementally =
+    !trimmed &&
+    (!currentLast ||
+      addedInOrder.every(
+        (event) => compareObserverEvents(event, currentLast) > 0,
+      ));
+  return { final, addedInOrder, canFoldIncrementally };
+}

desktop/src/features/agents/observerRelayStore.ts

@@ -21,12 +21,19 @@ import {
 import {
   compareObserverEvents,
   isObserverEventAfter,
+  mergeObserverEventBatch,
   unwrapObserverBatch,
 } from "./observerEventOrdering";
+import {
+  clearLatestLiveSessions,
+  getLatestLiveSessionId,
+  recordLatestLiveSession,
+} from "./latestLiveSession";
 // Re-export the channel-activity read and the event-ordering helpers off the
 // observer store's public surface; their storage/logic now lives in dedicated
 // modules but callers keep importing them from here.
 export { compareObserverEvents, getAgentChannelActivity, isObserverEventAfter };
+export { getLatestLiveSessionId };
 import { agentConfigSurfaceQueryKey } from "@/features/agents/hooks";
 import type {
   ConnectionState,
@@ -217,40 +224,9 @@ function truncateUnpinnedAgentWindow(key: string): void {
   notifyListeners();
 }
 
-// Per-agent, per-channel latest-live-session-id.
-// Key: `${normalizePubkey(agentPubkey)}:${channelId}`.
-// Set when a live relay observer event with a sessionId arrives.
-// Cleared in resetAgentObserverStore.
-//
-// "Latest-live" means: the sessionId that most recently appeared via the
-// live relay path (handleRelayObserverEvent). It is NOT derived from
-// connectionState or an ever-live Set — an ever-live Set would incorrectly
-// mark session A as "current" after session B has started (Thufir Pass 3).
-//
-// Stored as `{ sessionId, timestamp, seq }` so that late-arriving live frames
-// from an older session never regress the latest-live id. We only advance when
-// the parsed event sorts strictly AFTER the stored one, using the same
-// two-key ordering as `compareObserverEvents`: timestamp first, then seq on a
-// tie — so a higher-seq frame at equal timestamp still advances the entry.
-type LatestLiveEntry = { sessionId: string; timestamp: string; seq: number };
-const latestLiveSessionByAgentChannel = new Map<string, LatestLiveEntry>();
-
-function liveSessionKey(agentPubkey: string, channelId: string | null): string {
-  return `${normalizePubkey(agentPubkey)}:${channelId ?? ""}`;
-}
-
-/** Read the latest-live-session-id for a (agent, channel) pair. */
-export function getLatestLiveSessionId(
-  agentPubkey: string | null | undefined,
-  channelId: string | null | undefined,
-): string | null {
-  if (!agentPubkey) return null;
-  return (
-    latestLiveSessionByAgentChannel.get(
-      liveSessionKey(agentPubkey, channelId ?? null),
-    )?.sessionId ?? null
-  );
-}
+// Per-agent, per-channel latest-live-session-id lives in latestLiveSession.ts
+// (recorded on the live path, cleared on reset); its reader is re-exported off
+// the store's public surface below.
 
 // Per-agent listeners for `control_result` frames. The ModelPicker subscribes
 // here to learn the async outcome of a `switch_model` frame (the send is
@@ -359,46 +335,26 @@ function appendAgentEvents(
   if (events.length === 0) return false;
 
   const key = normalizePubkey(agentPubkey);
-  recordChannelActivity(key, event);
+  for (const event of events) recordChannelActivity(key, event);
   const current = eventsByAgent.get(key) ?? [];
-  const seen = new Set(
-    current.map(
-      (event) => `${event.timestamp.length}:${event.timestamp}:${event.seq}`,
-    ),
-  );
-  const added: ObserverEvent[] = [];
-  for (const event of events) {
-    const eventKey = `${event.timestamp.length}:${event.timestamp}:${event.seq}`;
-    if (seen.has(eventKey)) continue;
-    seen.add(eventKey);
-    added.push(event);
-  }
-  if (added.length === 0) return false;
+  const merged = mergeObserverEventBatch(current, events, agentEventCap(key));
+  if (!merged) return false;
 
-  const sortedAdded = added.sort(compareObserverEvents);
-  const sorted = [...current, ...sortedAdded].sort(compareObserverEvents);
-  const cap = agentEventCap(key);
-  const trimmed = sorted.length > cap;
-  const final = trimmed ? sorted.slice(sorted.length - cap) : sorted;
-  eventsByAgent.set(key, final);
+  eventsByAgent.set(key, merged.final);
 
   // The common live path appends a sorted batch after the retained window. Fold
   // that batch through the transcript state once without rebuilding history.
   // Out-of-order arrivals and cap eviction rebuild from the final window so
   // stateful tool/permission relationships remain correct.
-  const currentLast = current.at(-1);
-  const allAtEnd =
-    !currentLast ||
-    sortedAdded.every((event) => compareObserverEvents(event, currentLast) > 0);
-  if (allAtEnd && !trimmed) {
+  if (merged.canFoldIncrementally) {
     let transcriptState =
       transcriptByAgent.get(key) ?? createEmptyTranscriptState();
-    for (const event of sortedAdded) {
+    for (const event of merged.addedInOrder) {
       transcriptState = processTranscriptEvent(transcriptState, event);
     }
     transcriptByAgent.set(key, transcriptState);
   } else {
-    transcriptByAgent.set(key, buildTranscriptState(final));
+    transcriptByAgent.set(key, buildTranscriptState(merged.final));
   }
 
   invalidateSnapshot(key);
@@ -414,7 +370,7 @@ function appendAgentEvent(agentPubkey: string, event: ObserverEvent) {
 /**
  * Compose the map key for the channel-scoped archive transcript.
  * Separates agent identity from channel with `:` — the same delimiter used by
- * liveSessionKey so all composite keys in this module are consistently shaped.
+ * the latest-live-session key so composite keys are consistently shaped.
  */
 function archiveChannelKey(agentPubkey: string, channelId: string): string {
   return `${normalizePubkey(agentPubkey)}:${channelId}`;
@@ -526,25 +482,10 @@ function processLiveObserverEvents(
   const observerChanged = appendAgentEvents(agentPubkey, events);
 
   for (const parsed of events) {
-    // Track the latest-live-session-id per (agent, channel) on the live path.
-    // Only set when the parsed event carries both a sessionId and channelId,
-    // so we never attribute a session to the wrong channel.
-    if (parsed.sessionId && parsed.channelId) {
-      const key = liveSessionKey(agentPubkey, parsed.channelId);
-      const stored = latestLiveSessionByAgentChannel.get(key);
-      // Advance only when this event sorts strictly AFTER the stored one via
-      // isObserverEventAfter (timestamp then seq — same ordering as
-      // compareObserverEvents). This prevents late-arriving live frames from
-      // older sessions from regressing the latest-live id, while also
-      // correctly advancing on a same-timestamp frame with a higher seq.
-      if (!stored || isObserverEventAfter(parsed, stored)) {
-        latestLiveSessionByAgentChannel.set(key, {
-          sessionId: parsed.sessionId,
-          timestamp: parsed.timestamp,
-          seq: parsed.seq,
-        });
-      }
-    }
+    // Advance the latest-live-session-id for this (agent, channel) on the live
+    // path (no-op unless the frame carries a sessionId + channelId and sorts
+    // after the stored entry).
+    recordLatestLiveSession(agentPubkey, parsed);
     const managementRequest = parseAgentManagementRequest(parsed.payload);
     if (managementRequest) {
       for (const listener of agentManagementListeners) {
@@ -987,7 +928,7 @@ export function resetAgentObserverStore() {
   knownAgentPubkeys.clear();
   knownAgentsBySubscription.clear();
   pendingUnknownAgentFrames.length = 0;
-  latestLiveSessionByAgentChannel.clear();
+  clearLatestLiveSessions();
   agentManagementListeners.clear();
   onSessionConfigCaptured = null;
   connectionState = "idle";

desktop/src/features/messages/lib/projectChannelWindow.test.mjs

@@ -401,4 +401,3 @@ test("test_projection_aligns_an_existing_messages_key", () => {
 
   assert.deepEqual(contents(harness), ["initial", "live"]);
 });
-

desktop/src/features/profile/ui/UserProfilePopover.tsx

@@ -293,7 +293,6 @@ export function UserProfilePopover({
     viewerIsOwner,
   });
 
-
   React.useEffect(() => {
     return clearHoverTimer;
   }, [clearHoverTimer]);