diff --git a/desktop/src/features/agents/observerRelayStore.ts b/desktop/src/features/agents/observerRelayStore.ts index 6aa397edbc..1ac54e7efd 100644 --- a/desktop/src/features/agents/observerRelayStore.ts +++ b/desktop/src/features/agents/observerRelayStore.ts @@ -84,17 +84,26 @@ function registerKnownAgents( subscriptionId: string, pubkeys: readonly string[], ) { - knownAgentsBySubscription.set( - subscriptionId, - new Set(pubkeys.map((pubkey) => normalizePubkey(pubkey))), - ); + const normalizedPubkeys = pubkeys.map((pubkey) => normalizePubkey(pubkey)); + knownAgentsBySubscription.set(subscriptionId, new Set(normalizedPubkeys)); recomputeKnownAgentPubkeys(); + pillDiagLog("registerKnownAgents", { + subscriptionId, + subscriptionPubkeys: pillDiagPubkeys(normalizedPubkeys), + }); } function unregisterKnownAgents(subscriptionId: string) { - if (knownAgentsBySubscription.delete(subscriptionId)) { + const removedPubkeys = knownAgentsBySubscription.get(subscriptionId); + const removed = knownAgentsBySubscription.delete(subscriptionId); + if (removed) { recomputeKnownAgentPubkeys(); } + pillDiagLog("unregisterKnownAgents", { + subscriptionId, + removed, + removedPubkeys: removedPubkeys ? pillDiagPubkeys(removedPubkeys) : [], + }); } let connectionState: ConnectionState = "idle"; @@ -104,6 +113,44 @@ let startPromise: Promise | null = null; let eventProcessingQueue: Promise = Promise.resolve(); let generation = 0; +function pillDiagPubkeys(pubkeys: Iterable) { + return [...pubkeys].map((pubkey) => normalizePubkey(pubkey)).sort(); +} + +function pillDiagState(extra: Record = {}) { + return { + generation, + connectionState, + hasRelaySubscription: Boolean(unsubscribeRelay), + hasStartPromise: Boolean(startPromise), + knownAgentCount: knownAgentPubkeys.size, + knownAgents: pillDiagPubkeys(knownAgentPubkeys), + subscriptionCount: knownAgentsBySubscription.size, + subscriptionSizes: Object.fromEntries( + [...knownAgentsBySubscription.entries()].map( + ([subscriptionId, pubkeys]) => [subscriptionId, pubkeys.size], + ), + ), + at: new Date().toISOString(), + ...extra, + }; +} + +function pillDiagLog(label: string, extra: Record = {}) { + console.log(`[pill-diag] observer ${label}`, pillDiagState(extra)); +} + +function pillDiagEvent(event: RelayEvent) { + const agentPubkey = observerTag(event, "agent"); + return { + eventId: event.id, + eventPubkey: normalizePubkey(event.pubkey), + agentPubkey: agentPubkey ? normalizePubkey(agentPubkey) : null, + frame: observerTag(event, "frame"), + createdAt: event.created_at, + }; +} + function notifyListeners() { for (const listener of listeners) { listener(); @@ -191,26 +238,52 @@ async function handleRelayObserverEvent( const agentPubkey = observerTag(event, "agent"); const frame = observerTag(event, "frame"); if (!agentPubkey || frame !== "telemetry") { + pillDiagLog("drop nonTelemetryFrame", { + activeGeneration, + event: pillDiagEvent(event), + }); return; } // Verify agent is known/trusted before decrypting. // Silently drop events from agents we are not managing. if (!knownAgentPubkeys.has(normalizePubkey(agentPubkey))) { + pillDiagLog("drop unknownAgent", { + activeGeneration, + event: pillDiagEvent(event), + }); 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)) { + pillDiagLog("drop senderMismatch", { + activeGeneration, + event: pillDiagEvent(event), + }); return; } try { const parsed = (await decryptObserverEvent(event)) as ObserverEvent; if (activeGeneration !== generation) { + pillDiagLog("drop staleGenerationAfterDecrypt", { + activeGeneration, + event: pillDiagEvent(event), + parsedKind: parsed.kind, + parsedSeq: parsed.seq, + parsedTimestamp: parsed.timestamp, + }); return; } + pillDiagLog("appendFrame", { + activeGeneration, + event: pillDiagEvent(event), + parsedKind: parsed.kind, + parsedSeq: parsed.seq, + parsedTimestamp: parsed.timestamp, + }); appendAgentEvent(agentPubkey, parsed); if (parsed.kind === "session_config_captured") { void putAgentSessionConfig(agentPubkey, parsed.payload); @@ -220,8 +293,18 @@ async function handleRelayObserverEvent( } } catch (error) { if (activeGeneration !== generation) { + pillDiagLog("drop staleGenerationAfterError", { + activeGeneration, + event: pillDiagEvent(event), + error: error instanceof Error ? error.message : String(error), + }); return; } + pillDiagLog("decryptError", { + activeGeneration, + event: pillDiagEvent(event), + error: error instanceof Error ? error.message : String(error), + }); setConnectionState( "error", error instanceof Error @@ -233,25 +316,44 @@ async function handleRelayObserverEvent( export function ensureRelayObserverSubscription() { if (unsubscribeRelay) { + pillDiagLog("ensure reuseOpenSubscription"); return Promise.resolve(); } if (startPromise) { + pillDiagLog("ensure reuseStartPromise"); return startPromise; } const activeGeneration = generation; + pillDiagLog("ensure start", { activeGeneration }); setConnectionState("connecting", null); startPromise = (async () => { const identity = await getIdentity(); + pillDiagLog("subscribeToAgentObserverFrames start", { + activeGeneration, + identityPubkey: normalizePubkey(identity.pubkey), + }); const unsubscribe = await subscribeToAgentObserverFrames( identity.pubkey, (event) => { + pillDiagLog("frame received", { + activeGeneration, + event: pillDiagEvent(event), + }); eventProcessingQueue = eventProcessingQueue .then(() => handleRelayObserverEvent(event, activeGeneration)) .catch((error) => { if (activeGeneration !== generation) { + pillDiagLog("drop staleGenerationInQueueCatch", { + activeGeneration, + error: error instanceof Error ? error.message : String(error), + }); return; } + pillDiagLog("eventHandlingError", { + activeGeneration, + error: error instanceof Error ? error.message : String(error), + }); setConnectionState( "error", error instanceof Error @@ -262,13 +364,19 @@ export function ensureRelayObserverSubscription() { }, ); if (activeGeneration !== generation) { + pillDiagLog("subscribe resolved staleGeneration", { activeGeneration }); await unsubscribe(); return; } unsubscribeRelay = unsubscribe; + pillDiagLog("subscribe open", { activeGeneration }); setConnectionState("open", null); })() .catch((error) => { + pillDiagLog("subscribe error", { + activeGeneration, + error: error instanceof Error ? error.message : String(error), + }); if (activeGeneration === generation) { setConnectionState( "error", @@ -279,6 +387,7 @@ export function ensureRelayObserverSubscription() { } }) .finally(() => { + pillDiagLog("ensure finally", { activeGeneration }); if (activeGeneration === generation) { startPromise = null; } @@ -406,6 +515,7 @@ export function useManagedAgentObserverBridge( }, [subscriptionId, agentPubkeys]); React.useEffect(() => { + pillDiagLog("bridge activeEffect", { hasActiveAgent }); if (!hasActiveAgent) { return; } @@ -425,6 +535,7 @@ export function useManagedAgentObserverBridge( } export function resetAgentObserverStore() { + pillDiagLog("reset start", { nextGeneration: generation + 1 }); generation += 1; const unsubscribe = unsubscribeRelay; unsubscribeRelay = null; @@ -439,5 +550,6 @@ export function resetAgentObserverStore() { connectionState = "idle"; errorMessage = null; notifyListeners(); + pillDiagLog("reset afterClear", { hadUnsubscribe: Boolean(unsubscribe) }); void unsubscribe?.(); }