chore(sidebar): trace observer relay timing

Co-authored-by: Taylor Ho <taylorkmho@gmail.com>
Signed-off-by: Taylor Ho <taylorkmho@gmail.com>
This commit is contained in:
npub1223z34hd7vtwc6qj4s7flsxkj644nlre2nthu7lrrmkumhu3xddsrx9r6w
2026-06-30 20:42:26 -07:00
parent 8bbdd88a67
commit 4a1892a651

View File

@@ -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<void> | null = null;
let eventProcessingQueue: Promise<void> = Promise.resolve();
let generation = 0;
function pillDiagPubkeys(pubkeys: Iterable<string>) {
return [...pubkeys].map((pubkey) => normalizePubkey(pubkey)).sort();
}
function pillDiagState(extra: Record<string, unknown> = {}) {
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<string, unknown> = {}) {
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?.();
}