diff --git a/desktop/src/features/agents/observerRelayStore.ts b/desktop/src/features/agents/observerRelayStore.ts index 6d5d46f65..6aa397edb 100644 --- a/desktop/src/features/agents/observerRelayStore.ts +++ b/desktop/src/features/agents/observerRelayStore.ts @@ -84,26 +84,17 @@ function registerKnownAgents( subscriptionId: string, pubkeys: readonly string[], ) { - const normalizedPubkeys = pubkeys.map((pubkey) => normalizePubkey(pubkey)); - knownAgentsBySubscription.set(subscriptionId, new Set(normalizedPubkeys)); - recomputeKnownAgentPubkeys(); - pillDiagLog("registerKnownAgents", { + knownAgentsBySubscription.set( subscriptionId, - subscriptionPubkeys: pillDiagPubkeys(normalizedPubkeys), - }); + new Set(pubkeys.map((pubkey) => normalizePubkey(pubkey))), + ); + recomputeKnownAgentPubkeys(); } function unregisterKnownAgents(subscriptionId: string) { - const removedPubkeys = knownAgentsBySubscription.get(subscriptionId); - const removed = knownAgentsBySubscription.delete(subscriptionId); - if (removed) { + if (knownAgentsBySubscription.delete(subscriptionId)) { recomputeKnownAgentPubkeys(); } - pillDiagLog("unregisterKnownAgents", { - subscriptionId, - removed, - removedPubkeys: removedPubkeys ? pillDiagPubkeys(removedPubkeys) : [], - }); } let connectionState: ConnectionState = "idle"; @@ -113,54 +104,6 @@ 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 pillDiagBridgeAgents( - agents: readonly Pick[], -) { - return agents.map((agent) => ({ - pubkey: normalizePubkey(agent.pubkey), - status: agent.status, - startsObserver: agent.status === "running" || agent.status === "deployed", - })); -} - function notifyListeners() { for (const listener of listeners) { listener(); @@ -248,52 +191,26 @@ 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); @@ -303,18 +220,8 @@ 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 @@ -326,44 +233,25 @@ 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 @@ -374,19 +262,13 @@ 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", @@ -397,7 +279,6 @@ export function ensureRelayObserverSubscription() { } }) .finally(() => { - pillDiagLog("ensure finally", { activeGeneration }); if (activeGeneration === generation) { startPromise = null; } @@ -509,12 +390,6 @@ export function useManagedAgentObserverBridge( [agents], ); - pillDiagLog("bridge render", { - subscriptionId, - hasActiveAgent, - agents: pillDiagBridgeAgents(agents), - }); - const agentPubkeys = React.useMemo( () => agents.map((agent) => agent.pubkey), [agents], @@ -531,7 +406,6 @@ export function useManagedAgentObserverBridge( }, [subscriptionId, agentPubkeys]); React.useEffect(() => { - pillDiagLog("bridge activeEffect", { hasActiveAgent }); if (!hasActiveAgent) { return; } @@ -551,7 +425,6 @@ export function useManagedAgentObserverBridge( } export function resetAgentObserverStore() { - pillDiagLog("reset start", { nextGeneration: generation + 1 }); generation += 1; const unsubscribe = unsubscribeRelay; unsubscribeRelay = null; @@ -566,6 +439,5 @@ export function resetAgentObserverStore() { connectionState = "idle"; errorMessage = null; notifyListeners(); - pillDiagLog("reset afterClear", { hadUnsubscribe: Boolean(unsubscribe) }); void unsubscribe?.(); } diff --git a/desktop/src/features/sidebar/lib/useActiveWorkingChannelsById.test.mjs b/desktop/src/features/sidebar/lib/useActiveWorkingChannelsById.test.mjs index a13dc48d8..cbfc3718d 100644 --- a/desktop/src/features/sidebar/lib/useActiveWorkingChannelsById.test.mjs +++ b/desktop/src/features/sidebar/lib/useActiveWorkingChannelsById.test.mjs @@ -124,7 +124,7 @@ describe("getOwnedRelayWorkingAgents", () => { }); describe("mergeWorkingAgents", () => { - it("dedupes owned relay agents behind locally managed agents", () => { + it("keeps active managed agents ahead of owned relay duplicates", () => { assert.deepEqual( mergeWorkingAgents( [{ pubkey: "AAAA", name: "Ned", status: "running" }], @@ -143,4 +143,26 @@ describe("mergeWorkingAgents", () => { ], ); }); + + it("uses owned relay status when a duplicate managed agent is stopped", () => { + assert.deepEqual( + mergeWorkingAgents( + [ + { + pubkey: OWNED_RELAY_AGENT_PUBKEY.toUpperCase(), + name: "Local Nadia", + status: "stopped", + }, + ], + [ + { + pubkey: OWNED_RELAY_AGENT_PUBKEY, + name: "nadia", + status: "deployed", + }, + ], + ), + [{ pubkey: OWNED_RELAY_AGENT_PUBKEY, name: "nadia", status: "deployed" }], + ); + }); }); diff --git a/desktop/src/features/sidebar/lib/useActiveWorkingChannelsById.ts b/desktop/src/features/sidebar/lib/useActiveWorkingChannelsById.ts index d61658659..dc0de7f4d 100644 --- a/desktop/src/features/sidebar/lib/useActiveWorkingChannelsById.ts +++ b/desktop/src/features/sidebar/lib/useActiveWorkingChannelsById.ts @@ -20,17 +20,8 @@ import type { } from "@/shared/api/types"; import { normalizePubkey } from "@/shared/lib/pubkey"; -console.log("[pill-diag] module loaded", { - loadedAt: new Date().toISOString(), -}); - -let pillDiagRenderCount = 0; - type WorkingAgentName = Pick; type WorkingAgent = Pick; -type PillDiagProfile = UserProfileSummary & { - pubkey?: string; -}; type OwnedRelayWorkingAgent = Pick & { status: "deployed"; }; @@ -69,113 +60,35 @@ export function getOwnedRelayWorkingAgents( }); } +function agentCanStartObserver(agent: WorkingAgent) { + return agent.status === "running" || agent.status === "deployed"; +} + export function mergeWorkingAgents( managedAgents: readonly WorkingAgent[], ownedRelayAgents: readonly WorkingAgent[], ): WorkingAgent[] { - const seenPubkeys = new Set(); - const merged: WorkingAgent[] = []; + const mergedByPubkey = new Map(); - for (const agent of [...managedAgents, ...ownedRelayAgents]) { - const pubkey = normalizePubkey(agent.pubkey); - if (seenPubkeys.has(pubkey)) continue; - seenPubkeys.add(pubkey); - merged.push(agent); + for (const agent of managedAgents) { + mergedByPubkey.set(normalizePubkey(agent.pubkey), agent); } - return merged; -} + for (const agent of ownedRelayAgents) { + const pubkey = normalizePubkey(agent.pubkey); + const managedAgent = mergedByPubkey.get(pubkey); + if (!managedAgent || !agentCanStartObserver(managedAgent)) { + mergedByPubkey.set(pubkey, agent); + } + } -function summarizeAgent(agent: WorkingAgentName & { status?: string }) { - return { - pubkey: normalizePubkey(agent.pubkey), - name: agent.name, - status: agent.status ?? null, - }; -} - -function summarizeRelayProfile( - pubkey: string, - profile: PillDiagProfile | undefined, - currentPubkey: string | undefined, -) { - return { - pubkey: normalizePubkey(pubkey), - profileKeyPubkey: profile?.pubkey ? normalizePubkey(profile.pubkey) : null, - displayName: profile?.displayName ?? null, - isAgent: profile?.isAgent ?? null, - ownerPubkey: profile?.ownerPubkey - ? normalizePubkey(profile.ownerPubkey) - : null, - ownsAuthorAgent: ownsAuthorAgent(profile, currentPubkey), - }; -} - -function logPillDiagnostics({ - currentPubkey, - managedAgents, - relayAgents, - profiles, - ownedRelayAgents, - workingAgents, - activeWorkingChannels, -}: { - currentPubkey: string | undefined; - managedAgents: readonly WorkingAgent[]; - relayAgents: readonly Pick[]; - profiles: Record | undefined; - ownedRelayAgents: readonly WorkingAgent[]; - workingAgents: readonly WorkingAgent[]; - activeWorkingChannels: readonly ActiveChannelTurnSummary[]; -}) { - const normalizedProfiles = Object.fromEntries( - relayAgents.map((agent) => { - const pubkey = normalizePubkey(agent.pubkey); - const profile = profiles?.[pubkey] as PillDiagProfile | undefined; - return [ - pubkey, - summarizeRelayProfile(agent.pubkey, profile, currentPubkey), - ]; - }), - ); - - console.groupCollapsed("[pill-diag] active working channels inputs", { - currentPubkey: currentPubkey ? normalizePubkey(currentPubkey) : null, - managedAgentCount: managedAgents.length, - relayAgentCount: relayAgents.length, - ownedRelayAgentCount: ownedRelayAgents.length, - workingAgentCount: workingAgents.length, - activeChannelCount: activeWorkingChannels.length, - }); - console.log("[pill-diag] currentPubkey", { - currentPubkey: currentPubkey ? normalizePubkey(currentPubkey) : null, - }); - console.table(managedAgents.map(summarizeAgent)); - console.log("[pill-diag] managedAgents", managedAgents.map(summarizeAgent)); - console.table(relayAgents.map(summarizeAgent)); - console.log("[pill-diag] relayAgents", relayAgents.map(summarizeAgent)); - console.log("[pill-diag] profilesByRelayAgent", normalizedProfiles); - console.table(ownedRelayAgents.map(summarizeAgent)); - console.log( - "[pill-diag] ownedRelayAgents", - ownedRelayAgents.map(summarizeAgent), - ); - console.table(workingAgents.map(summarizeAgent)); - console.log("[pill-diag] workingAgents", workingAgents.map(summarizeAgent)); - console.log("[pill-diag] activeWorkingChannels", activeWorkingChannels); - console.groupEnd(); + return [...mergedByPubkey.values()]; } export function useActiveWorkingChannelsById(): ReadonlyMap< string, ActiveChannelTurnSummary > { - pillDiagRenderCount += 1; - console.log("[pill-diag] hook render", { - renderCount: pillDiagRenderCount, - renderedAt: new Date().toISOString(), - }); - const identityQuery = useIdentityQuery(); const currentPubkey = identityQuery.data?.pubkey; const managedAgentsQuery = useManagedAgentsQuery(); @@ -214,26 +127,6 @@ export function useActiveWorkingChannelsById(): ReadonlyMap< const activeWorkingChannels = useActiveAgentTurnsByChannel(); - React.useEffect(() => { - logPillDiagnostics({ - currentPubkey, - managedAgents, - relayAgents, - profiles: relayAgentProfilesQuery.data?.profiles, - ownedRelayAgents, - workingAgents, - activeWorkingChannels, - }); - }, [ - activeWorkingChannels, - currentPubkey, - managedAgents, - ownedRelayAgents, - relayAgentProfilesQuery.data?.profiles, - relayAgents, - workingAgents, - ]); - return React.useMemo( () => new Map(