chore(desktop): diagnose silently dropped observer frames

desktop/src/features/agents/ingestArchivedObserverEvents.test.mjs

@@ -18,6 +18,7 @@ import {
   resetAgentObserverStore,
   _testRegisterKnownAgents,
   _testGetArchivedChannelEvents,
+  _testGetObserverDropCounts,
 } from "@/features/agents/observerRelayStore.ts";
 
 // ── Constants ─────────────────────────────────────────────────────────────────
@@ -129,6 +130,11 @@ describe("ingestArchivedObserverEvents", () => {
     });
     await ingestArchivedObserverEvents([nonTelemetryEvent], decryptFn);
     assert.equal(decryptCalled, false, "non-telemetry frame must be dropped");
+    assert.equal(
+      _testGetObserverDropCounts().missing_telemetry_tag,
+      undefined,
+      "valid non-telemetry frames must not emit drop diagnostics",
+    );
     const snap = getAgentObserverSnapshot(AGENT_PUBKEY, true);
     assert.equal(snap.events.length, 0);
   });
@@ -142,6 +148,36 @@ describe("ingestArchivedObserverEvents", () => {
     assert.equal(snap.events.length, 0);
   });
 
+  it("aggregates observer drop diagnostics by stable reason", async () => {
+    _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]);
+    await ingestArchivedObserverEvents(
+      [
+        makeRawEvent({
+          tags: [["agent", AGENT_PUBKEY]],
+        }),
+        makeRawEvent({
+          tags: [
+            ["agent", OTHER_PUBKEY],
+            ["frame", "telemetry"],
+          ],
+        }),
+        makeRawEvent({
+          pubkey: OTHER_PUBKEY,
+        }),
+        makeRawEvent(),
+        makeRawEvent(),
+      ],
+      makeDecryptFail(),
+    );
+
+    assert.deepEqual(_testGetObserverDropCounts(), {
+      missing_telemetry_tag: 1,
+      unknown_agent: 1,
+      sender_agent_mismatch: 1,
+      decrypt_failed: 2,
+    });
+  });
+
   it("test_successful_ingest_adds_event_to_store", async () => {
     _testRegisterKnownAgents(SUB_ID, [AGENT_PUBKEY]);
     const obs = makeObserverEvent({ seq: 1 });

desktop/src/features/agents/observerRelayStore.ts

@@ -32,6 +32,19 @@ import {
 
 const MAX_OBSERVER_EVENTS = 3000;
 const MAX_PENDING_UNKNOWN_AGENT_FRAMES = 100;
+const OBSERVER_DROP_LOG_INTERVAL_MS = 10_000;
+
+type ObserverDropReason =
+  | "missing_telemetry_tag"
+  | "unknown_agent"
+  | "sender_agent_mismatch"
+  | "stale_generation"
+  | "decrypt_failed";
+
+type ObserverDropLogState = {
+  count: number;
+  lastLoggedAt: number;
+};
 
 export type ObserverSnapshot = {
   connectionState: ConnectionState;
@@ -120,6 +133,43 @@ const agentManagementListeners = new Set<
 const knownAgentPubkeys = new Set<string>();
 const knownAgentsBySubscription = new Map<string, Set<string>>();
 const pendingUnknownAgentFrames: RelayEvent[] = [];
+const observerDropLogState = new Map<
+  ObserverDropReason,
+  ObserverDropLogState
+>();
+
+function logObserverDrop(
+  reason: ObserverDropReason,
+  event: RelayEvent,
+  activeGeneration: number,
+) {
+  const previous = observerDropLogState.get(reason) ?? {
+    count: 0,
+    lastLoggedAt: 0,
+  };
+  const count = previous.count + 1;
+  const now = Date.now();
+  const shouldLog =
+    count === 1 ||
+    now - previous.lastLoggedAt >= OBSERVER_DROP_LOG_INTERVAL_MS ||
+    count % 100 === 0;
+  observerDropLogState.set(reason, {
+    count,
+    lastLoggedAt: shouldLog ? now : previous.lastLoggedAt,
+  });
+  if (!shouldLog) return;
+
+  const agentTag = observerTag(event, "agent");
+  console.debug("[observerRelayStore] observer frame dropped", {
+    reason,
+    count,
+    eventId: event.id,
+    agentTag,
+    senderPubkey: event.pubkey,
+    currentGeneration: generation,
+    eventGeneration: activeGeneration,
+  });
+}
 
 // Callback invoked when session_config_captured is received, so React Query
 // can invalidate the config-surface query for the affected agent. Wired up
@@ -349,9 +399,11 @@ async function handleRelayObserverEvent(
 ) {
   const agentPubkey = observerTag(event, "agent");
   const frame = observerTag(event, "frame");
-  if (!agentPubkey || frame !== "telemetry") {
+  if (!agentPubkey || frame == null) {
+    logObserverDrop("missing_telemetry_tag", event, activeGeneration);
     return;
   }
+  if (frame !== "telemetry") return;
 
   // Ownership data arrives asynchronously during startup. Buffer raw signed
   // frames until the first trusted-agent set is registered, then re-run this
@@ -362,19 +414,23 @@ async function handleRelayObserverEvent(
       if (pendingUnknownAgentFrames.length > MAX_PENDING_UNKNOWN_AGENT_FRAMES) {
         pendingUnknownAgentFrames.shift();
       }
+    } else {
+      logObserverDrop("unknown_agent", event, activeGeneration);
     }
     return;
   }
 
   // Defense-in-depth: verify the event sender matches the claimed agent pubkey.
   // The relay gates on is_agent_owner, but a compromised relay could misroute.
   if (normalizePubkey(event.pubkey) !== normalizePubkey(agentPubkey)) {
+    logObserverDrop("sender_agent_mismatch", event, activeGeneration);
     return;
   }
 
   try {
     const parsed = (await decryptObserverEvent(event)) as ObserverEvent;
     if (activeGeneration !== generation) {
+      logObserverDrop("stale_generation", event, activeGeneration);
       return;
     }
     // Track the latest-live-session-id per (agent, channel) on the live path.
@@ -417,8 +473,10 @@ async function handleRelayObserverEvent(
     }
   } catch (error) {
     if (activeGeneration !== generation) {
+      logObserverDrop("stale_generation", event, activeGeneration);
       return;
     }
+    logObserverDrop("decrypt_failed", event, activeGeneration);
     setConnectionState(
       "error",
       error instanceof Error
@@ -670,13 +728,19 @@ export async function ingestArchivedObserverEvents(
   for (const event of rawEvents) {
     const agentPubkey = observerTag(event, "agent");
     const frame = observerTag(event, "frame");
-    if (!agentPubkey || frame !== "telemetry") {
+    if (!agentPubkey || frame == null) {
+      logObserverDrop("missing_telemetry_tag", event, generation);
       continue;
     }
+    if (frame !== "telemetry") continue;
     if (!knownAgentPubkeys.has(normalizePubkey(agentPubkey))) {
+      if (knownAgentPubkeys.size > 0) {
+        logObserverDrop("unknown_agent", event, generation);
+      }
       continue;
     }
     if (normalizePubkey(event.pubkey) !== normalizePubkey(agentPubkey)) {
+      logObserverDrop("sender_agent_mismatch", event, generation);
       continue;
     }
     try {
@@ -697,7 +761,7 @@ export async function ingestArchivedObserverEvents(
         appendAgentEvent(agentPubkey, parsed);
       }
     } catch {
-      // Silently drop decrypt failures — same as live path error handling.
+      logObserverDrop("decrypt_failed", event, generation);
     }
   }
   // Batch-notify once for the whole page of archive events. appendAgentEvent
@@ -754,6 +818,7 @@ export function resetAgentObserverStore() {
   knownAgentPubkeys.clear();
   knownAgentsBySubscription.clear();
   pendingUnknownAgentFrames.length = 0;
+  observerDropLogState.clear();
   latestLiveSessionByAgentChannel.clear();
   agentManagementListeners.clear();
   onSessionConfigCaptured = null;
@@ -788,3 +853,15 @@ export function _testGetArchivedChannelEvents(
     archiveEventsByChannel.get(archiveChannelKey(agentPubkey, channelId)) ?? []
   );
 }
+
+/** Test-only: read the aggregated live observer drop counters. */
+export function _testGetObserverDropCounts(): Record<
+  ObserverDropReason,
+  number
+> {
+  const counts = {} as Record<ObserverDropReason, number>;
+  for (const [reason, state] of observerDropLogState) {
+    counts[reason] = state.count;
+  }
+  return counts;
+}