Batch observer-store publications per relay envelope (#5680)

## Summary - preserve the ACP observer envelope through renderer ingestion - bulk-deduplicate/sort/fold one agent batch before one external-store publication - suppress publications for entirely duplicate replay batches - cover raw history, transcript, active-turn terminal behavior, and publication count ## Why The harness already publishes observer frames in one-second batches. Desktop expanded each envelope and called the global observer store once per inner frame. Each call copied/sorted up to 3,000 retained frames and woke every observer subscriber; the app-level active-turn bridge then rescanned every running/deployed agent's retained buffer. ## Representative work-count profile Controlled workload: 14 agents, 1,000 retained frames each, 24 inner frames/envelope, 10 rounds (3,360 new frames). | Counter | Before | After | |---|---:|---:| | Observer publications | 3,360 | 140 | | Aggregate retained events revisited by a representative global subscriber | 52,686,480 | 2,196,880 | Both deterministic counters fall **24×**. Node wall time was loader/JIT-noisy and is deliberately not presented as production CPU evidence. ## Validation Exact head `038a29f6f0ff866884e07bb66eebe87e576f6769`: - `pnpm --dir desktop test` — 4,718 passed, 0 failed - `pnpm --dir desktop typecheck` — passed before rebase; the rebase changed only the base and the full suite passed on the exact head - pre-commit Desktop Biome + file-size gate — passed The installed v0.5.10-block process and LocalStorage database were not restarted or modified. Signed-off-by: Wes <wesbillman@users.noreply.github.com> Co-authored-by: Princess Donut <68157ebd23b3897c1991015c3038658ea916200c67d3a54620b0754d1b92f6e0@buzz.block.builderlab.xyz>
2026-08-18 06:50:31 +02:00 · 2026-08-12 11:29:19 -07:00
parent dc2dbfe0f5
commit c3b0ccf383
2 changed files with 218 additions and 65 deletions
@@ -16,7 +16,11 @@ import {
 import {
  injectObserverEventsForE2E,
  getAgentObserverSnapshot,
+  getAgentTranscript,
+  subscribeAgentObserverStore,
+  subscribeAgentManagementRequests,
  resetAgentObserverStore,
+  _testProcessLiveObserverEvents,
 } from "./observerRelayStore.ts";
 import { formatElapsed } from "./ui/agentSessionUtils.ts";

@@ -1621,6 +1625,115 @@ describe("observer → active-turns bridge sync", () => {
    );
  });

+  it("publishes one observer update for a batch while preserving outcomes", () => {
+    let observerNotifications = 0;
+    const unsubscribeObserver = subscribeAgentObserverStore(() => {
+      observerNotifications += 1;
+    });
+    const events = [
+      makeEvent({ seq: 1, kind: "turn_started" }),
+      ...Array.from({ length: 98 }, (_, index) =>
+        makeEvent({
+          seq: index + 2,
+          kind: "acp_write",
+          timestamp: new Date(
+            Date.parse("2024-01-01T00:00:00Z") + (index + 1) * 1_000,
+          ).toISOString(),
+        }),
+      ),
+      makeEvent({
+        seq: 100,
+        kind: "turn_completed",
+        timestamp: "2024-01-01T00:02:00Z",
+      }),
+    ];
+
+    injectObserverEventsForE2E(AGENT, events);
+    unsubscribeObserver();
+
+    assert.equal(
+      observerNotifications,
+      1,
+      "one harness envelope must produce one external-store publication",
+    );
+    assert.equal(getAgentObserverSnapshot(AGENT, true).events.length, 100);
+    assert.ok(
+      getAgentTranscript(AGENT, true).length > 0,
+      "the batched events must still build transcript state",
+    );
+    syncActiveAgentTurnsFromObserver(bridgeAgents);
+    assert.equal(
+      getActiveTurnsForAgent(AGENT).length,
+      0,
+      "the terminal event must still clear derived liveness",
+    );
+  });
+
+  it("makes the triggering frame visible before management callbacks", () => {
+    const managementFrame = makeEvent({
+      seq: 2,
+      kind: "acp_message",
+      timestamp: "2024-01-01T00:00:01Z",
+      payload: {
+        type: "agent_management_request",
+        action: "create",
+        requestId: "request-1",
+        request: {
+          channelId: "chan-1",
+          displayName: "Fleet Observer",
+          systemPrompt: "Observe the fleet.",
+        },
+      },
+    });
+    let visibleSeqs = [];
+    let observerNotifications = 0;
+    const unsubscribeObserver = subscribeAgentObserverStore(() => {
+      observerNotifications += 1;
+    });
+    const unsubscribeManagement = subscribeAgentManagementRequests(
+      (agentPubkey) => {
+        assert.equal(agentPubkey, AGENT);
+        visibleSeqs = getAgentObserverSnapshot(AGENT, true).events.map(
+          (event) => event.seq,
+        );
+        assert.equal(
+          observerNotifications,
+          0,
+          "the global publication must remain deferred until callbacks finish",
+        );
+      },
+    );
+
+    _testProcessLiveObserverEvents(AGENT, [
+      makeEvent({ seq: 1, kind: "turn_started" }),
+      managementFrame,
+    ]);
+    unsubscribeManagement();
+    unsubscribeObserver();
+
+    assert.deepEqual(
+      visibleSeqs,
+      [1, 2],
+      "management callback must observe its triggering frame and prior envelope frames",
+    );
+    assert.equal(observerNotifications, 1);
+  });
+
+  it("does not publish when a replay batch is entirely duplicate", () => {
+    const events = [makeEvent({ seq: 1, kind: "turn_started" })];
+    injectObserverEventsForE2E(AGENT, events);
+
+    let observerNotifications = 0;
+    const unsubscribeObserver = subscribeAgentObserverStore(() => {
+      observerNotifications += 1;
+    });
+    injectObserverEventsForE2E(AGENT, events);
+    unsubscribeObserver();
+
+    assert.equal(observerNotifications, 0);
+    assert.equal(getAgentObserverSnapshot(AGENT, true).events.length, 1);
+  });
+
  it("skips agents that are neither running nor deployed", () => {
    injectObserverEventsForE2E(AGENT, [
      makeEvent({ seq: 1, kind: "turn_started" }),
@@ -193,44 +193,63 @@ function observerTag(event: RelayEvent, tagName: string) {
  return event.tags.find((tag) => tag[0] === tagName)?.[1] ?? null;
 }

-function appendAgentEvent(agentPubkey: string, event: ObserverEvent) {
+function appendAgentEvents(
+  agentPubkey: string,
+  events: readonly ObserverEvent[],
+): boolean {
+  if (events.length === 0) return false;
+
  const key = normalizePubkey(agentPubkey);
  const current = eventsByAgent.get(key) ?? [];
-  if (
-    current.some(
-      (existing) =>
-        existing.seq === event.seq && existing.timestamp === event.timestamp,
-    )
-  ) {
-    return;
+  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 sorted = [...current, event].sort(compareObserverEvents);
+  const sortedAdded = added.sort(compareObserverEvents);
+  const sorted = [...current, ...sortedAdded].sort(compareObserverEvents);
  const trimmed = sorted.length > MAX_OBSERVER_EVENTS;
  const final = trimmed
    ? sorted.slice(sorted.length - MAX_OBSERVER_EVENTS)
    : sorted;
  eventsByAgent.set(key, final);

-  // Determine whether the new event landed at the end of the sorted array.
-  // If it did (common case), we can incrementally process just this event.
-  // If not (out-of-order arrival) or if we trimmed, fall back to full rebuild.
-  const eventAtEnd = sorted[sorted.length - 1] === event;
-
-  if (eventAtEnd && !trimmed) {
-    // Fast path: incremental update
-    const transcriptState =
+  // 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) {
+    let transcriptState =
      transcriptByAgent.get(key) ?? createEmptyTranscriptState();
-    const updatedTranscript = processTranscriptEvent(transcriptState, event);
-    transcriptByAgent.set(key, updatedTranscript);
+    for (const event of sortedAdded) {
+      transcriptState = processTranscriptEvent(transcriptState, event);
+    }
+    transcriptByAgent.set(key, transcriptState);
  } else {
-    // Slow path: full rebuild (out-of-order insertion or trim fired)
    transcriptByAgent.set(key, buildTranscriptState(final));
  }

  invalidateSnapshot(key);
+  return true;
+}

-  notifyListeners();
+function appendAgentEvent(agentPubkey: string, event: ObserverEvent) {
+  if (appendAgentEvents(agentPubkey, [event])) {
+    notifyListeners();
+  }
 }

 /**
@@ -362,44 +381,60 @@ function unwrapObserverBatch(parsed: ObserverEvent): ObserverEvent[] {

 // Per-event processing shared by every event a live frame carries (one for a
 // plain frame, many for a batch envelope).
-function processLiveObserverEvent(agentPubkey: string, parsed: ObserverEvent) {
-  // 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,
-      });
+function processLiveObserverEvents(
+  agentPubkey: string,
+  events: readonly ObserverEvent[],
+) {
+  // Commit the full envelope before dispatching synchronous specialized
+  // callbacks. Those callbacks historically observed their triggering frame
+  // in the raw/transcript stores; batching must preserve that visibility while
+  // deferring only the global external-store publication.
+  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,
+        });
+      }
+    }
+    const managementRequest = parseAgentManagementRequest(parsed.payload);
+    if (managementRequest) {
+      for (const listener of agentManagementListeners) {
+        listener(agentPubkey, managementRequest);
+      }
+    }
+    if (parsed.kind === "session_config_captured") {
+      void putAgentSessionConfig(agentPubkey, parsed.payload);
+      onSessionConfigCaptured?.(agentPubkey);
+    } else if (parsed.kind === "control_result") {
+      dispatchControlResult(agentPubkey, parsed.payload);
+    } else if (parsed.kind === "managed_agent_runtime_lifecycle") {
+      void putManagedAgentRuntimeLifecycle(agentPubkey, parsed.payload).catch(
+        (error) => {
+          console.debug("Late/untracked lifecycle frame dropped:", error);
+        },
+      );
    }
  }
-  appendAgentEvent(agentPubkey, parsed);
-  const managementRequest = parseAgentManagementRequest(parsed.payload);
-  if (managementRequest) {
-    for (const listener of agentManagementListeners) {
-      listener(agentPubkey, managementRequest);
-    }
-  }
-  if (parsed.kind === "session_config_captured") {
-    void putAgentSessionConfig(agentPubkey, parsed.payload);
-    onSessionConfigCaptured?.(agentPubkey);
-  } else if (parsed.kind === "control_result") {
-    dispatchControlResult(agentPubkey, parsed.payload);
-  } else if (parsed.kind === "managed_agent_runtime_lifecycle") {
-    void putManagedAgentRuntimeLifecycle(agentPubkey, parsed.payload).catch(
-      (error) => {
-        console.debug("Late/untracked lifecycle frame dropped:", error);
-      },
-    );
+
+  // Preserve the harness's envelope backpressure: retained state was committed
+  // before specialized callbacks, but external-store subscribers publish once.
+  if (observerChanged) {
+    notifyListeners();
  }
 }

@@ -437,9 +472,7 @@ async function handleRelayObserverEvent(
    if (activeGeneration !== generation) {
      return;
    }
-    for (const inner of unwrapObserverBatch(parsed)) {
-      processLiveObserverEvent(agentPubkey, inner);
-    }
+    processLiveObserverEvents(agentPubkey, unwrapObserverBatch(parsed));
  } catch (error) {
    if (activeGeneration !== generation) {
      return;
@@ -748,10 +781,9 @@ export function injectObserverEventsForE2E(
  agentPubkey: string,
  events: ObserverEvent[],
 ) {
-  for (const event of events) {
-    appendAgentEvent(agentPubkey, event);
+  if (appendAgentEvents(agentPubkey, events)) {
+    notifyListeners();
  }
-  notifyListeners();
 }

 /**
@@ -762,8 +794,8 @@ export function syncAgentObserverEvents(
  agentPubkey: string,
  events: ObserverEvent[],
 ) {
-  for (const event of events) {
-    appendAgentEvent(agentPubkey, event);
+  if (appendAgentEvents(agentPubkey, events)) {
+    notifyListeners();
  }
 }

@@ -801,6 +833,14 @@ export function _testRegisterKnownAgents(
  registerKnownAgents(subscriptionId, pubkeys);
 }

+/** Test-only: exercise live envelope ordering without relay/decryption setup. */
+export function _testProcessLiveObserverEvents(
+  agentPubkey: string,
+  events: readonly ObserverEvent[],
+): void {
+  processLiveObserverEvents(agentPubkey, events);
+}
+
 /**
 * Test-only: read the raw archived observer events for a (agent, channel) pair.
 * Production callers should use `getArchivedChannelEvents`.