diff --git a/.gitignore b/.gitignore index f26e74136c0..69a484bacf6 100644 --- a/.gitignore +++ b/.gitignore @@ -72,3 +72,6 @@ identity.key # Helm dependency tarballs — regenerable from Chart.lock via `helm dependency build` deploy/charts/*/charts/*.tgz + +# Claude Code context +.claude_context_tree diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 1352b31cad8..0f89521ac27 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1426,6 +1426,14 @@ struct SlotCircuit { /// Prevents duplicate spawns from maintenance ticks that fire before the /// previous spawn_and_init completes. respawn_in_flight: bool, + /// True once a circuit-open alert has been emitted for this slot and + /// not yet followed by a successful respawn. Lets the respawn-complete + /// path emit a matching "recovered" alert exactly once. + alerted_open: bool, + /// Channel of the prompt that triggered the crash behind `alerted_open`, + /// if any. Carried forward so the "recovered" alert renders in the same + /// channel as the "suspended" alert it resolves, instead of nowhere. + alerted_channel_id: Option, } /// Result of [`SlotCircuit::record_crash`]. @@ -1807,6 +1815,8 @@ mod idle_pool_sleep_tests { crash_times: Vec::new(), open_until: None, respawn_in_flight, + alerted_open: false, + alerted_channel_id: None, } } @@ -2373,6 +2383,8 @@ async fn tokio_main() -> Result<()> { crash_times: Vec::new(), open_until: None, respawn_in_flight: false, + alerted_open: false, + alerted_channel_id: None, }) .collect(); @@ -2487,6 +2499,11 @@ async fn tokio_main() -> Result<()> { }; pool.return_agent(agent); tracing::info!(agent = rr.index, "respawn complete"); + if crash_history[rr.index].alerted_open { + crash_history[rr.index].alerted_open = false; + let channel_id = crash_history[rr.index].alerted_channel_id.take(); + emit_circuit_recovered_alert(observer.as_ref(), rr.index, channel_id); + } respawn_collected = true; } Err(e) => { @@ -3851,6 +3868,74 @@ fn spawn_failure_notice( } } +/// Emit a user-visible alert when an agent slot's circuit breaker opens. +/// +/// Previously this state transition (3 crashes/60s, or a failed half-open +/// probe re-opening the circuit) only produced a `tracing::error!` log — +/// invisible unless someone is watching server logs, so a permanently dark +/// slot could go unnoticed indefinitely. This routes through the same +/// owner-encrypted observer-frame pipeline already used for `agent_panic` +/// and `turn_error`; the desktop app renders it via a dedicated +/// `circuit_open`/`circuit_recovered` branch in `processTranscriptEvent` +/// (desktop/src/features/agents/ui/agentSessionTranscript.ts) — no new +/// transport, but client-side rendering had to be added alongside this. +fn emit_circuit_open_alert( + observer: Option<&observer::ObserverHandle>, + agent_index: usize, + channel_id: Option, + trigger: &str, +) { + let Some(observer) = observer else { + return; + }; + let cooldown_secs = CIRCUIT_BREAKER_COOLDOWN.as_secs(); + observer.emit( + "circuit_open", + Some(agent_index), + &observer::context_for(channel_id, None, None), + serde_json::json!({ + "trigger": trigger, + "cooldown_secs": cooldown_secs, + "error": format!( + "Agent slot {agent_index} {trigger} repeatedly and its circuit breaker is now open \ + — it will not respond until the {cooldown_secs}s cooldown elapses and a health probe succeeds." + ), + }), + ); +} + +/// Emit a user-visible alert when a slot that previously tripped its circuit +/// breaker ([`emit_circuit_open_alert`]) has successfully respawned. +/// +/// Closes the loop for an owner watching the observer stream: they see both +/// when a slot went dark and when it came back, rather than only the former. +/// +/// `channel_id` is the channel of the prompt that triggered the crash behind +/// this recovery, if any — carried from `SlotCircuit::alerted_channel_id` so +/// this alert renders in the same channel as the "suspended" alert it +/// resolves. Previously this was always emitted with no channel, so it had no +/// channel-scoped surface in the desktop app and the "suspended" message +/// never appeared to resolve. +fn emit_circuit_recovered_alert( + observer: Option<&observer::ObserverHandle>, + agent_index: usize, + channel_id: Option, +) { + let Some(observer) = observer else { + return; + }; + observer.emit( + "circuit_recovered", + Some(agent_index), + &observer::context_for(channel_id, None, None), + serde_json::json!({ + "error": format!( + "Agent slot {agent_index} recovered — its circuit breaker probe succeeded and it is responding again." + ), + }), + ); +} + #[allow(clippy::too_many_arguments)] fn handle_prompt_result( pool: &mut AgentPool, @@ -3867,6 +3952,14 @@ fn handle_prompt_result( ) -> LoopAction { let before = pool.task_map().len(); let agent_index = result.agent.index; + // The channel that triggered this prompt, if any — threaded through to + // any crash-recovery alert below so it has somewhere to render instead + // of silently emitting with no channel context (a heartbeat prompt has + // no channel, so None is correct there). + let source_channel_id = match result.source { + PromptSource::Channel(channel_id) => Some(channel_id), + PromptSource::Heartbeat => None, + }; let successful_steer_deliveries = pool .task_map() .values() @@ -4112,6 +4205,7 @@ fn handle_prompt_result( respawn_tx, respawn_tasks, observer.clone(), + source_channel_id, ) { // Circuit open — slot stays empty until maintenance refill. if pool.live_count() == 0 && !any_respawn_in_flight(crash_history) { @@ -4152,6 +4246,7 @@ fn handle_prompt_result( respawn_tx, respawn_tasks, observer.clone(), + source_channel_id, ) { // Circuit open — slot stays empty until maintenance refill. if pool.live_count() == 0 && !any_respawn_in_flight(crash_history) { @@ -4217,6 +4312,7 @@ fn handle_prompt_result( respawn_tx, respawn_tasks, observer, + source_channel_id, ) && pool.live_count() == 0 && !any_respawn_in_flight(crash_history) { @@ -4307,6 +4403,9 @@ fn recover_panicked_agent( let delay = match slot.record_crash() { CrashVerdict::CircuitOpen => { tracing::error!(agent = i, "circuit open after panic — not respawning"); + emit_circuit_open_alert(observer.as_ref(), i, meta.channel_id, "panicked"); + slot.alerted_open = true; + slot.alerted_channel_id = meta.channel_id; return; } CrashVerdict::HalfOpenProbe => { @@ -4506,6 +4605,13 @@ fn default_heartbeat_prompt() -> String { /// the actual shutdown + backoff + spawn_and_init work into a background task. /// The result comes back through `respawn_tx` so the main loop stays responsive. /// +/// `channel_id` is the channel of the prompt that triggered this crash, if +/// any (`None` for a heartbeat-triggered crash) — threaded through so a +/// circuit-open alert has somewhere to render; previously this was always +/// `None` here, so most circuit-open alerts (the generic crash/timeout path, +/// as opposed to the panic path which already had `meta.channel_id`) had no +/// channel-scoped surface in the desktop app. +/// /// Returns `true` if a respawn task was spawned, `false` if the circuit is open. fn spawn_respawn_task( old_agent: OwnedAgent, @@ -4514,6 +4620,7 @@ fn spawn_respawn_task( respawn_tx: &mpsc::Sender, respawn_tasks: &mut tokio::task::JoinSet<()>, observer: Option, + channel_id: Option, ) -> bool { let index = old_agent.index; @@ -4521,6 +4628,9 @@ fn spawn_respawn_task( let delay = match slot.record_crash() { CrashVerdict::CircuitOpen => { tracing::error!(agent = index, "circuit open — not respawning"); + emit_circuit_open_alert(observer.as_ref(), index, channel_id, "crashed"); + slot.alerted_open = true; + slot.alerted_channel_id = channel_id; return false; } CrashVerdict::HalfOpenProbe => { @@ -7096,6 +7206,8 @@ mod error_outcome_emission_tests { crash_times: Vec::new(), open_until: None, respawn_in_flight: false, + alerted_open: false, + alerted_channel_id: None, }]; let (respawn_tx, _respawn_rx) = mpsc::channel(8); let mut respawn_tasks = tokio::task::JoinSet::new(); @@ -7168,6 +7280,8 @@ mod error_outcome_emission_tests { crash_times: Vec::new(), open_until: None, respawn_in_flight: false, + alerted_open: false, + alerted_channel_id: None, }]; let (respawn_tx, _respawn_rx) = mpsc::channel(8); let mut respawn_tasks = tokio::task::JoinSet::new(); @@ -7282,6 +7396,8 @@ mod error_outcome_emission_tests { crash_times: Vec::new(), open_until: None, respawn_in_flight: false, + alerted_open: false, + alerted_channel_id: None, }]; let (respawn_tx, _respawn_rx) = mpsc::channel(8); let mut respawn_tasks = tokio::task::JoinSet::new(); @@ -7343,6 +7459,8 @@ mod error_outcome_emission_tests { crash_times: Vec::new(), open_until: None, respawn_in_flight: false, + alerted_open: false, + alerted_channel_id: None, }]; let (respawn_tx, _respawn_rx) = mpsc::channel(8); let mut respawn_tasks = tokio::task::JoinSet::new(); @@ -7424,6 +7542,8 @@ mod error_outcome_emission_tests { crash_times: Vec::new(), open_until: None, respawn_in_flight: false, + alerted_open: false, + alerted_channel_id: None, }]; let (respawn_tx, _respawn_rx) = mpsc::channel(8); let mut respawn_tasks = tokio::task::JoinSet::new(); @@ -7512,6 +7632,8 @@ mod error_outcome_emission_tests { crash_times: Vec::new(), open_until: None, respawn_in_flight: false, + alerted_open: false, + alerted_channel_id: None, }]; let (respawn_tx, _respawn_rx) = mpsc::channel(8); let mut respawn_tasks = tokio::task::JoinSet::new(); @@ -7604,6 +7726,8 @@ mod error_outcome_emission_tests { crash_times: Vec::new(), open_until: None, respawn_in_flight: false, + alerted_open: false, + alerted_channel_id: None, }]; let (respawn_tx, _respawn_rx) = mpsc::channel(8); let mut respawn_tasks = tokio::task::JoinSet::new(); @@ -7710,6 +7834,8 @@ mod error_outcome_emission_tests { crash_times: Vec::new(), open_until: None, respawn_in_flight: false, + alerted_open: false, + alerted_channel_id: None, }]; let (respawn_tx, _respawn_rx) = mpsc::channel(8); let mut respawn_tasks = tokio::task::JoinSet::new(); @@ -7787,6 +7913,8 @@ mod error_outcome_emission_tests { crash_times: Vec::new(), open_until: None, respawn_in_flight: false, + alerted_open: false, + alerted_channel_id: None, }]; let (respawn_tx, _respawn_rx) = mpsc::channel(8); let mut respawn_tasks = tokio::task::JoinSet::new(); @@ -7881,6 +8009,8 @@ mod error_outcome_emission_tests { crash_times: Vec::new(), open_until: None, respawn_in_flight: false, + alerted_open: false, + alerted_channel_id: None, }]; let (respawn_tx, _respawn_rx) = mpsc::channel(8); let mut respawn_tasks = tokio::task::JoinSet::new(); @@ -8009,6 +8139,8 @@ mod error_outcome_emission_tests { crash_times: Vec::new(), open_until: None, respawn_in_flight: false, + alerted_open: false, + alerted_channel_id: None, }]; let (respawn_tx, _respawn_rx) = mpsc::channel(8); let mut respawn_tasks = tokio::task::JoinSet::new(); @@ -8139,6 +8271,8 @@ mod error_outcome_emission_tests { crash_times: Vec::new(), open_until: None, respawn_in_flight: false, + alerted_open: false, + alerted_channel_id: None, }]; let (respawn_tx, _respawn_rx) = mpsc::channel(8); let mut respawn_tasks = tokio::task::JoinSet::new(); @@ -8328,6 +8462,8 @@ mod error_outcome_emission_tests { crash_times: Vec::new(), open_until: None, respawn_in_flight: false, + alerted_open: false, + alerted_channel_id: None, }]; let (respawn_tx, _respawn_rx) = mpsc::channel(8); let mut respawn_tasks = tokio::task::JoinSet::new(); @@ -8414,6 +8550,8 @@ mod error_outcome_emission_tests { crash_times: Vec::new(), open_until: None, respawn_in_flight: false, + alerted_open: false, + alerted_channel_id: None, }]; let (respawn_tx, _respawn_rx) = mpsc::channel(8); let mut respawn_tasks = tokio::task::JoinSet::new(); diff --git a/desktop/playwright.config.ts b/desktop/playwright.config.ts index b1a2d5623b2..73cc1dcec19 100644 --- a/desktop/playwright.config.ts +++ b/desktop/playwright.config.ts @@ -151,6 +151,7 @@ export default defineConfig({ "**/huddle-transcription.spec.ts", "**/agent-numeric-tuning.spec.ts", "**/needs-restart-screenshots.spec.ts", + "**/circuit-open-badge-screenshots.spec.ts", ], use: { ...devices["Desktop Chrome"], diff --git a/desktop/src/app/routes/WorkflowsRouteScreen.tsx b/desktop/src/app/routes/WorkflowsRouteScreen.tsx index 193695f0cd2..35ce3bd4f41 100644 --- a/desktop/src/app/routes/WorkflowsRouteScreen.tsx +++ b/desktop/src/app/routes/WorkflowsRouteScreen.tsx @@ -33,11 +33,18 @@ export function WorkflowsRouteScreen({ }, [editor?.hasOrigin, goWorkflows]); const channelsQuery = useChannelsQuery(); const channels = channelsQuery.data ?? []; - const memberChannels = channels.filter((channel) => channel.isMember); + // Member channels plus open channels the owner hasn't joined yet — open + // channels are readable without membership (see channelDescription.ts), + // so a workflow living there is fully visible and must not be silently + // dropped just because isMember is false. WorkflowsView narrows further + // for membership-gated actions (e.g. the create-workflow channel picker). + const visibleChannels = channels.filter( + (channel) => channel.isMember || channel.visibility === "open", + ); return ( { diff --git a/desktop/src/features/agents/agentCircuitHooks.ts b/desktop/src/features/agents/agentCircuitHooks.ts new file mode 100644 index 00000000000..dd9881b9ee2 --- /dev/null +++ b/desktop/src/features/agents/agentCircuitHooks.ts @@ -0,0 +1,55 @@ +//! Circuit-status React hooks, split out of `observerRelayStore.ts`. +//! +//! Merging main pushed that file past the repo's 1000-line budget, and an +//! over-budget file may not grow. These two hooks are the cleanest seam: they +//! read circuit state from `agentCircuitStatus` and use the store only for its +//! subscribe function, so nothing module-private moves with them. +//! +//! They live here rather than in `agentCircuitStatus.ts` because that module is +//! imported by `observerRelayStore`; depending on the store from there would +//! close an import cycle. + +import * as React from "react"; + +import { normalizePubkey } from "@/shared/lib/pubkey"; +import { + getAgentCircuitStatus, + getOpenCircuitPubkeySignature, + type AgentCircuitStatus, +} from "./agentCircuitStatus"; +import { subscribeAgentObserverStore } from "./observerRelayStore"; + +/** Persistent per-agent circuit-breaker status, independent of channel/transcript scope. */ +export function useAgentCircuitStatus( + agentPubkey: string | null | undefined, +): AgentCircuitStatus { + return React.useSyncExternalStore(subscribeAgentObserverStore, () => + getAgentCircuitStatus(agentPubkey), + ); +} + +/** + * The subset of `agents` whose circuit is currently open. Generic over the + * caller's own agent shape so it works with any `{ pubkey: string }`-shaped + * roster (e.g. a channel's bot list) without this module needing to know + * about it. The `useSyncExternalStore` snapshot is a primitive signature + * string (see `getOpenCircuitPubkeySignature`) rather than an array, so it's + * reference-stable across renders where nothing changed without this module + * needing a cache; the actual `T[]` is then derived per-render via a normal + * `useMemo` keyed on both `agents` and that signature. + */ +export function useOpenCircuitAgents( + agents: readonly T[], +): T[] { + const signature = React.useSyncExternalStore( + subscribeAgentObserverStore, + () => getOpenCircuitPubkeySignature(agents.map((agent) => agent.pubkey)), + ); + return React.useMemo(() => { + if (!signature) return []; + const openPubkeys = new Set(signature.split(",")); + return agents.filter((agent) => + openPubkeys.has(normalizePubkey(agent.pubkey)), + ); + }, [agents, signature]); +} diff --git a/desktop/src/features/agents/agentCircuitStatus.ts b/desktop/src/features/agents/agentCircuitStatus.ts new file mode 100644 index 00000000000..0838b970340 --- /dev/null +++ b/desktop/src/features/agents/agentCircuitStatus.ts @@ -0,0 +1,185 @@ +import { normalizePubkey } from "@/shared/lib/pubkey"; +import type { ObserverEvent } from "./ui/agentSessionTypes"; +import { isObserverEventAfter } from "./lib/observerEventOrdering"; + +export type AgentCircuitStatus = { + isOpen: boolean; + message: string | null; + channelId: string | null; + timestamp: string | null; + /** Backend cooldown window in seconds (`CIRCUIT_BREAKER_COOLDOWN`), from the + * circuit_open event's payload. Anchored at `timestamp`, so the caller + * computes remaining time itself — see `circuitCooldownRemainingMs`. Null + * when unknown (e.g. a circuit_recovered event, which carries no cooldown). */ + cooldownSecs: number | null; +}; + +const IDLE_CIRCUIT_STATUS: AgentCircuitStatus = { + isOpen: false, + message: null, + channelId: null, + timestamp: null, + cooldownSecs: null, +}; + +// Latest known circuit-breaker state per (agent, slot). Slots are independent +// prompt lanes within one managed-agent process (see SlotCircuit in +// buzz-acp/src/lib.rs) — a recovery on slot 1 must not clear a still-open +// circuit on slot 0, so state is tracked per slot and the derived per-agent +// status reports "open" if ANY slot is open. +type CircuitSlotState = { + isOpen: boolean; + message: string; + channelId: string | null; + timestamp: string; + seq: number; + cooldownSecs: number | null; +}; +const circuitStateBySlot = new Map(); + +// Reference-stable per-agent snapshots for useSyncExternalStore, invalidated +// only when that agent's circuit state actually changes. +const circuitStatusCache = new Map(); + +function circuitSlotKey( + agentPubkey: string, + agentIndex: number | null | undefined, +): string { + return `${normalizePubkey(agentPubkey)}:${agentIndex ?? 0}`; +} + +/** + * Fold a circuit_open/circuit_recovered event into per-slot circuit state. + * Ignores any other event kind and any event that doesn't sort strictly after + * the slot's currently-stored state, so a late/replayed frame can't regress + * an already-newer status. Returns true if state actually changed. + */ +export function applyCircuitEvent( + agentPubkey: string, + event: ObserverEvent, +): boolean { + if (event.kind !== "circuit_open" && event.kind !== "circuit_recovered") { + return false; + } + const key = circuitSlotKey(agentPubkey, event.agentIndex); + const existing = circuitStateBySlot.get(key); + if (existing && !isObserverEventAfter(event, existing)) { + return false; + } + const payload = event.payload as { + error?: unknown; + cooldown_secs?: unknown; + } | null; + const message = + typeof payload?.error === "string" + ? payload.error + : event.kind === "circuit_open" + ? "Agent suspended (repeated crashes)" + : "Agent recovered"; + const cooldownSecs = + typeof payload?.cooldown_secs === "number" && + Number.isFinite(payload.cooldown_secs) + ? payload.cooldown_secs + : null; + circuitStateBySlot.set(key, { + isOpen: event.kind === "circuit_open", + message, + channelId: event.channelId ?? null, + timestamp: event.timestamp, + seq: event.seq, + cooldownSecs, + }); + circuitStatusCache.delete(normalizePubkey(agentPubkey)); + return true; +} + +/** + * Derived circuit-breaker status for one agent: open if ANY of its slots is + * currently open. Reference-stable while unchanged (useSyncExternalStore-safe + * — see the React.memo/useSyncExternalStore gotcha in this repo's CLAUDE.md). + */ +export function getAgentCircuitStatus( + agentPubkey?: string | null, +): AgentCircuitStatus { + if (!agentPubkey) { + return IDLE_CIRCUIT_STATUS; + } + const key = normalizePubkey(agentPubkey); + const cached = circuitStatusCache.get(key); + if (cached) { + return cached; + } + const prefix = `${key}:`; + let openSlot: CircuitSlotState | null = null; + for (const [slotKey, slot] of circuitStateBySlot) { + if (!slotKey.startsWith(prefix) || !slot.isOpen) continue; + if (!openSlot || isObserverEventAfter(slot, openSlot)) { + openSlot = slot; + } + } + const status: AgentCircuitStatus = openSlot + ? { + isOpen: true, + message: openSlot.message, + channelId: openSlot.channelId, + timestamp: openSlot.timestamp, + cooldownSecs: openSlot.cooldownSecs, + } + : IDLE_CIRCUIT_STATUS; + circuitStatusCache.set(key, status); + return status; +} + +/** + * Comma-joined, sorted normalized pubkeys (from the given candidates) whose + * circuit is currently open. A primitive string rather than an array so + * callers can drive `useSyncExternalStore` directly with it — string equality + * gives reference-stability for free (`Object.is` on equal strings), with no + * cache to invalidate or leak, unlike returning a fresh array each call would + * require. Callers needing the actual agent objects re-derive them from this + * signature with a plain `useMemo` (see `BotActivityComposerAction`). + */ +export function getOpenCircuitPubkeySignature( + agentPubkeys: readonly string[], +): string { + return agentPubkeys + .filter((pubkey) => getAgentCircuitStatus(pubkey).isOpen) + .map(normalizePubkey) + .sort() + .join(","); +} + +/** + * Milliseconds remaining in the circuit's cooldown window, or null when not + * open or the cooldown is unknown. Clamped to 0 rather than negative once the + * window has elapsed — callers use that to distinguish "still cooling down" + * from "cooldown elapsed, health probe should be underway" without needing to + * re-derive the sign themselves. Pure function of `status` and `nowMs` so the + * caller supplies its own ticking clock (e.g. `useNow`) rather than this + * module owning a timer. + */ +export function circuitCooldownRemainingMs( + status: AgentCircuitStatus, + nowMs: number, +): number | null { + if ( + !status.isOpen || + status.timestamp == null || + status.cooldownSecs == null + ) { + return null; + } + const openedAtMs = Date.parse(status.timestamp); + if (!Number.isFinite(openedAtMs)) { + return null; + } + const remaining = openedAtMs + status.cooldownSecs * 1000 - nowMs; + return remaining > 0 ? remaining : 0; +} + +/** Clears all per-agent circuit state. Called from resetAgentObserverStore so + * circuit state can't leak across community switches. */ +export function resetCircuitState() { + circuitStateBySlot.clear(); + circuitStatusCache.clear(); +} diff --git a/desktop/src/features/agents/lib/observerEventOrdering.ts b/desktop/src/features/agents/lib/observerEventOrdering.ts new file mode 100644 index 00000000000..476538c34fc --- /dev/null +++ b/desktop/src/features/agents/lib/observerEventOrdering.ts @@ -0,0 +1,45 @@ +import type { ObserverEvent } from "../ui/agentSessionTypes"; + +/** + * Shared two-key ordering for observer events: later timestamp wins; equal + * timestamp falls back to higher seq. Extracted out of observerRelayStore.ts + * so other per-agent derived stores (e.g. agentCircuitStatus.ts) can apply the + * exact same ordering without importing the whole observer store module and + * without drifting from it. + */ +export function compareObserverEvents( + left: ObserverEvent, + right: ObserverEvent, +) { + const leftTime = Date.parse(left.timestamp); + const rightTime = Date.parse(right.timestamp); + if (Number.isFinite(leftTime) && Number.isFinite(rightTime)) { + const timeDiff = leftTime - rightTime; + if (timeDiff !== 0) { + return timeDiff; + } + } + + return left.seq - right.seq; +} + +/** + * Returns true if `candidate` sorts strictly after `stored` using the same + * two-key ordering as `compareObserverEvents`: later timestamp wins; equal + * timestamp falls back to higher seq. Extracted so latest-live advancement + * (and any other derived per-agent state) cannot drift from transcript + * ordering. + */ +export function isObserverEventAfter( + candidate: { timestamp: string; seq: number }, + stored: { timestamp: string; seq: number }, +): boolean { + const candidateTime = Date.parse(candidate.timestamp); + const storedTime = Date.parse(stored.timestamp); + if (Number.isFinite(candidateTime) && Number.isFinite(storedTime)) { + if (candidateTime !== storedTime) { + return candidateTime > storedTime; + } + } + return candidate.seq > stored.seq; +} diff --git a/desktop/src/features/agents/observerRelayStore.circuitStatus.test.mjs b/desktop/src/features/agents/observerRelayStore.circuitStatus.test.mjs new file mode 100644 index 00000000000..57bf5e92200 --- /dev/null +++ b/desktop/src/features/agents/observerRelayStore.circuitStatus.test.mjs @@ -0,0 +1,538 @@ +import assert from "node:assert/strict"; +import { describe, it, beforeEach, afterEach } from "node:test"; + +import { + circuitCooldownRemainingMs, + getAgentCircuitStatus, + getOpenCircuitPubkeySignature, + injectObserverEventsForE2E, + resetAgentObserverStore, + subscribeAgentObserverStore, + syncAgentObserverEvents, + _testProcessLiveObserverEvents, +} from "./observerRelayStore.ts"; + +const AGENT = + "abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234"; +const AGENT_2 = + "dcba4321dcba4321dcba4321dcba4321dcba4321dcba4321dcba4321dcba4321"; + +function makeEvent(overrides) { + return { + seq: 1, + timestamp: "2024-01-01T00:00:00Z", + kind: "circuit_open", + agentIndex: 0, + channelId: "chan-1", + sessionId: null, + turnId: null, + payload: null, + ...overrides, + }; +} + +describe("getAgentCircuitStatus", () => { + beforeEach(() => { + resetAgentObserverStore(); + }); + + afterEach(() => { + resetAgentObserverStore(); + }); + + it("reports isOpen false when no circuit events have been seen for the agent", () => { + const status = getAgentCircuitStatus(AGENT); + assert.equal(status.isOpen, false); + assert.equal(status.message, null); + assert.equal(status.channelId, null); + assert.equal(status.timestamp, null); + assert.equal(status.cooldownSecs, null); + }); + + it("flips isOpen true on a circuit_open event, surfacing its message/channel/timestamp/cooldown", () => { + _testProcessLiveObserverEvents(AGENT, [ + makeEvent({ + seq: 1, + timestamp: "2024-01-01T00:00:00Z", + kind: "circuit_open", + agentIndex: 0, + channelId: "chan-uuid-1", + payload: { + error: + "Agent slot 0 panicked repeatedly and its circuit breaker is now open.", + cooldown_secs: 300, + }, + }), + ]); + + const status = getAgentCircuitStatus(AGENT); + assert.equal(status.isOpen, true); + assert.equal( + status.message, + "Agent slot 0 panicked repeatedly and its circuit breaker is now open.", + ); + assert.equal(status.channelId, "chan-uuid-1"); + assert.equal(status.timestamp, "2024-01-01T00:00:00Z"); + assert.equal(status.cooldownSecs, 300); + }); + + it("leaves cooldownSecs null when the payload doesn't carry a numeric cooldown_secs", () => { + _testProcessLiveObserverEvents(AGENT, [ + makeEvent({ + seq: 1, + timestamp: "2024-01-01T00:00:00Z", + kind: "circuit_open", + agentIndex: 0, + channelId: "chan-uuid-1", + payload: { error: "opened", cooldown_secs: "not-a-number" }, + }), + ]); + + assert.equal(getAgentCircuitStatus(AGENT).cooldownSecs, null); + }); + + it("flips isOpen back to false when a newer circuit_recovered event arrives on the same slot", () => { + _testProcessLiveObserverEvents(AGENT, [ + makeEvent({ + seq: 1, + timestamp: "2024-01-01T00:00:00Z", + kind: "circuit_open", + agentIndex: 0, + channelId: "chan-uuid-1", + payload: { error: "opened" }, + }), + ]); + assert.equal(getAgentCircuitStatus(AGENT).isOpen, true); + + _testProcessLiveObserverEvents(AGENT, [ + makeEvent({ + seq: 2, + timestamp: "2024-01-01T00:01:00Z", + kind: "circuit_recovered", + agentIndex: 0, + channelId: "chan-uuid-1", + payload: { error: "recovered" }, + }), + ]); + + const status = getAgentCircuitStatus(AGENT); + assert.equal(status.isOpen, false); + assert.equal(status.message, null); + assert.equal(status.channelId, null); + assert.equal(status.timestamp, null); + }); + + it("reports isOpen true overall (any-slot-open) with two independent slots, one open and one recovered", () => { + // Slot 0 stays open. + _testProcessLiveObserverEvents(AGENT, [ + makeEvent({ + seq: 1, + timestamp: "2024-01-01T00:00:00Z", + kind: "circuit_open", + agentIndex: 0, + channelId: "chan-slot-0", + payload: { error: "slot 0 opened" }, + }), + ]); + // Slot 1 opens then recovers. + _testProcessLiveObserverEvents(AGENT, [ + makeEvent({ + seq: 2, + timestamp: "2024-01-01T00:00:01Z", + kind: "circuit_open", + agentIndex: 1, + channelId: "chan-slot-1", + payload: { error: "slot 1 opened" }, + }), + makeEvent({ + seq: 3, + timestamp: "2024-01-01T00:00:02Z", + kind: "circuit_recovered", + agentIndex: 1, + channelId: "chan-slot-1", + payload: { error: "slot 1 recovered" }, + }), + ]); + + const status = getAgentCircuitStatus(AGENT); + assert.equal( + status.isOpen, + true, + "any slot still open must report the agent as open", + ); + assert.equal(status.channelId, "chan-slot-0"); + assert.equal(status.message, "slot 0 opened"); + }); + + it("ignores a stale/out-of-order event arriving after a recovery", () => { + _testProcessLiveObserverEvents(AGENT, [ + makeEvent({ + seq: 1, + timestamp: "2024-01-01T00:00:00Z", + kind: "circuit_open", + agentIndex: 0, + channelId: "chan-uuid-1", + payload: { error: "opened" }, + }), + makeEvent({ + seq: 2, + timestamp: "2024-01-01T00:01:00Z", + kind: "circuit_recovered", + agentIndex: 0, + channelId: "chan-uuid-1", + payload: { error: "recovered" }, + }), + ]); + assert.equal(getAgentCircuitStatus(AGENT).isOpen, false); + + // A delayed circuit_open with an OLDER timestamp+seq than the currently + // stored (recovered) state must be ignored — it must not reopen the + // circuit or otherwise change status. + _testProcessLiveObserverEvents(AGENT, [ + makeEvent({ + seq: 1, + timestamp: "2024-01-01T00:00:00Z", + kind: "circuit_open", + agentIndex: 0, + channelId: "chan-uuid-1", + payload: { error: "stale reopen" }, + }), + ]); + + const status = getAgentCircuitStatus(AGENT); + assert.equal( + status.isOpen, + false, + "a stale out-of-order event must not regress an already-newer status", + ); + assert.equal(status.message, null); + assert.equal(status.channelId, null); + assert.equal(status.timestamp, null); + }); + + it("returns a stable reference across repeated calls with no intervening event", () => { + _testProcessLiveObserverEvents(AGENT, [ + makeEvent({ + seq: 1, + timestamp: "2024-01-01T00:00:00Z", + kind: "circuit_open", + agentIndex: 0, + channelId: "chan-uuid-1", + payload: { error: "opened" }, + }), + ]); + + const ref1 = getAgentCircuitStatus(AGENT); + const ref2 = getAgentCircuitStatus(AGENT); + assert.strictEqual( + ref1, + ref2, + "must return the cached object reference when nothing changed — a " + + "regression here would cause an infinite useSyncExternalStore render loop", + ); + }); + + it("notifies subscribers on a circuit change even when appendAgentEvents rejects the raw event as a dedup collision", () => { + // appendAgentEvents dedups purely on (timestamp, seq), independent of + // kind — so a circuit_open event that happens to share its (timestamp, + // seq) with an already-appended, unrelated event is treated as a + // duplicate by the raw journal (appendAgentEvents returns false) even + // though it is the first-ever circuit event for its own slot, so + // applyCircuitEvent legitimately applies it (returns true). Regression + // guard for the case where only appendAgentEvents' return value gated + // notifyListeners(): the circuit state would update silently with no + // subscriber ever told, so useAgentCircuitStatus consumers (the + // suspended-agent badge) would not re-render until unrelated traffic + // happened to touch the same agent. + _testProcessLiveObserverEvents(AGENT, [ + makeEvent({ + seq: 7, + timestamp: "2024-01-01T00:00:00Z", + kind: "acp_read", + agentIndex: null, + channelId: "chan-uuid-1", + payload: {}, + }), + ]); + + let notifyCount = 0; + const unsubscribe = subscribeAgentObserverStore(() => { + notifyCount += 1; + }); + try { + _testProcessLiveObserverEvents(AGENT, [ + makeEvent({ + seq: 7, + timestamp: "2024-01-01T00:00:00Z", + kind: "circuit_open", + agentIndex: 2, + channelId: "chan-uuid-2", + payload: { error: "slot 2 opened" }, + }), + ]); + } finally { + unsubscribe(); + } + + assert.equal( + notifyCount, + 1, + "a circuit-only state change must still notify useSyncExternalStore subscribers", + ); + assert.equal(getAgentCircuitStatus(AGENT).isOpen, true); + }); + + it("resetAgentObserverStore clears circuit state back to isOpen:false", () => { + _testProcessLiveObserverEvents(AGENT, [ + makeEvent({ + seq: 1, + timestamp: "2024-01-01T00:00:00Z", + kind: "circuit_open", + agentIndex: 0, + channelId: "chan-uuid-1", + payload: { error: "opened" }, + }), + ]); + assert.equal(getAgentCircuitStatus(AGENT).isOpen, true); + + resetAgentObserverStore(); + + const status = getAgentCircuitStatus(AGENT); + assert.equal(status.isOpen, false); + assert.equal(status.message, null); + assert.equal(status.channelId, null); + assert.equal(status.timestamp, null); + }); +}); + +describe("injectObserverEventsForE2E and syncAgentObserverEvents fold circuit events", () => { + // Regression coverage: a prior merge with upstream (which changed + // appendAgentEvents' return shape) silently dropped the applyCircuitEvent + // loop from both of these functions — no textual conflict, so it merged + // clean but wrong. _testProcessLiveObserverEvents alone wouldn't have + // caught it, since neither of these functions routes through it. + beforeEach(() => { + resetAgentObserverStore(); + }); + + afterEach(() => { + resetAgentObserverStore(); + }); + + it("injectObserverEventsForE2E updates circuit status, not just the raw journal", () => { + injectObserverEventsForE2E(AGENT, [ + makeEvent({ + seq: 1, + timestamp: "2024-01-01T00:00:00Z", + kind: "circuit_open", + agentIndex: 0, + channelId: "chan-uuid-1", + payload: { error: "opened" }, + }), + ]); + + assert.equal(getAgentCircuitStatus(AGENT).isOpen, true); + }); + + it("syncAgentObserverEvents updates circuit status, not just the raw journal", () => { + syncAgentObserverEvents(AGENT, [ + makeEvent({ + seq: 1, + timestamp: "2024-01-01T00:00:00Z", + kind: "circuit_open", + agentIndex: 0, + channelId: "chan-uuid-1", + payload: { error: "opened" }, + }), + ]); + + assert.equal(getAgentCircuitStatus(AGENT).isOpen, true); + }); + + it("injectObserverEventsForE2E notifies subscribers on a circuit-only change (no new raw events appended)", () => { + injectObserverEventsForE2E(AGENT, [ + makeEvent({ + seq: 1, + timestamp: "2024-01-01T00:00:00Z", + kind: "acp_read", + agentIndex: null, + channelId: "chan-uuid-1", + payload: {}, + }), + ]); + + let notifyCount = 0; + const unsubscribe = subscribeAgentObserverStore(() => { + notifyCount += 1; + }); + try { + // Same (timestamp, seq) as the event above — appendAgentEvents treats + // this as a duplicate and returns null, but it's the first-ever circuit + // event for this slot, so applyCircuitEvent still applies it. + injectObserverEventsForE2E(AGENT, [ + makeEvent({ + seq: 1, + timestamp: "2024-01-01T00:00:00Z", + kind: "circuit_open", + agentIndex: 0, + channelId: "chan-uuid-1", + payload: { error: "opened" }, + }), + ]); + } finally { + unsubscribe(); + } + + assert.equal(notifyCount, 1); + assert.equal(getAgentCircuitStatus(AGENT).isOpen, true); + }); +}); + +describe("circuitCooldownRemainingMs", () => { + const OPEN_STATUS = { + isOpen: true, + message: "opened", + channelId: "chan-1", + timestamp: "2024-01-01T00:00:00.000Z", + cooldownSecs: 300, + }; + const OPENED_AT_MS = Date.parse(OPEN_STATUS.timestamp); + + it("returns the full window right when the circuit opens", () => { + assert.equal( + circuitCooldownRemainingMs(OPEN_STATUS, OPENED_AT_MS), + 300_000, + ); + }); + + it("counts down as time passes", () => { + assert.equal( + circuitCooldownRemainingMs(OPEN_STATUS, OPENED_AT_MS + 120_000), + 180_000, + ); + }); + + it("clamps to 0 once the cooldown window has elapsed, never negative", () => { + assert.equal( + circuitCooldownRemainingMs(OPEN_STATUS, OPENED_AT_MS + 300_000), + 0, + ); + assert.equal( + circuitCooldownRemainingMs(OPEN_STATUS, OPENED_AT_MS + 999_000), + 0, + ); + }); + + it("returns null when the circuit isn't open", () => { + assert.equal( + circuitCooldownRemainingMs( + { ...OPEN_STATUS, isOpen: false }, + OPENED_AT_MS, + ), + null, + ); + }); + + it("returns null when cooldownSecs is unknown", () => { + assert.equal( + circuitCooldownRemainingMs( + { ...OPEN_STATUS, cooldownSecs: null }, + OPENED_AT_MS, + ), + null, + ); + }); + + it("returns null when timestamp is unknown or unparseable", () => { + assert.equal( + circuitCooldownRemainingMs( + { ...OPEN_STATUS, timestamp: null }, + OPENED_AT_MS, + ), + null, + ); + assert.equal( + circuitCooldownRemainingMs( + { ...OPEN_STATUS, timestamp: "not-a-date" }, + OPENED_AT_MS, + ), + null, + ); + }); +}); + +describe("getOpenCircuitPubkeySignature", () => { + beforeEach(() => { + resetAgentObserverStore(); + }); + + afterEach(() => { + resetAgentObserverStore(); + }); + + it("returns an empty string when none of the candidates are open", () => { + assert.equal(getOpenCircuitPubkeySignature([AGENT, AGENT_2]), ""); + }); + + it("includes only the open agent's normalized pubkey", () => { + _testProcessLiveObserverEvents(AGENT, [ + makeEvent({ + seq: 1, + timestamp: "2024-01-01T00:00:00Z", + kind: "circuit_open", + agentIndex: 0, + payload: { error: "opened" }, + }), + ]); + + assert.equal( + getOpenCircuitPubkeySignature([AGENT, AGENT_2]), + AGENT.toLowerCase(), + ); + }); + + it("sorts multiple open pubkeys so the signature is stable regardless of input order", () => { + for (const [agent, index] of [ + [AGENT, 0], + [AGENT_2, 0], + ]) { + _testProcessLiveObserverEvents(agent, [ + makeEvent({ + seq: 1, + timestamp: "2024-01-01T00:00:00Z", + kind: "circuit_open", + agentIndex: index, + payload: { error: "opened" }, + }), + ]); + } + + const expected = [AGENT, AGENT_2] + .map((a) => a.toLowerCase()) + .sort() + .join(","); + assert.equal(getOpenCircuitPubkeySignature([AGENT, AGENT_2]), expected); + assert.equal(getOpenCircuitPubkeySignature([AGENT_2, AGENT]), expected); + }); + + it("excludes candidates whose circuit has recovered", () => { + _testProcessLiveObserverEvents(AGENT, [ + makeEvent({ + seq: 1, + timestamp: "2024-01-01T00:00:00Z", + kind: "circuit_open", + agentIndex: 0, + payload: { error: "opened" }, + }), + makeEvent({ + seq: 2, + timestamp: "2024-01-01T00:01:00Z", + kind: "circuit_recovered", + agentIndex: 0, + payload: { error: "recovered" }, + }), + ]); + + assert.equal(getOpenCircuitPubkeySignature([AGENT, AGENT_2]), ""); + }); +}); diff --git a/desktop/src/features/agents/observerRelayStore.ts b/desktop/src/features/agents/observerRelayStore.ts index 68fa290ad25..173c468931c 100644 --- a/desktop/src/features/agents/observerRelayStore.ts +++ b/desktop/src/features/agents/observerRelayStore.ts @@ -25,6 +25,22 @@ import { createEmptyTranscriptState, processTranscriptEvent, } from "./ui/agentSessionTranscript"; +import { + compareObserverEvents, + isObserverEventAfter, +} from "./lib/observerEventOrdering"; +import { applyCircuitEvent, resetCircuitState } from "./agentCircuitStatus"; + +export { + compareObserverEvents, + isObserverEventAfter, +} from "./lib/observerEventOrdering"; +export { + circuitCooldownRemainingMs, + getAgentCircuitStatus, + getOpenCircuitPubkeySignature, + type AgentCircuitStatus, +} from "./agentCircuitStatus"; const MAX_OBSERVER_EVENTS = 3000; // Length the per-agent journal is evicted down to when it overflows @@ -400,42 +416,6 @@ export function getArchivedChannelEvents( ); } -export function compareObserverEvents( - left: ObserverEvent, - right: ObserverEvent, -) { - const leftTime = Date.parse(left.timestamp); - const rightTime = Date.parse(right.timestamp); - if (Number.isFinite(leftTime) && Number.isFinite(rightTime)) { - const timeDiff = leftTime - rightTime; - if (timeDiff !== 0) { - return timeDiff; - } - } - - return left.seq - right.seq; -} - -/** - * Returns true if `candidate` sorts strictly after `stored` using the same - * two-key ordering as `compareObserverEvents`: later timestamp wins; equal - * timestamp falls back to higher seq. Extracted so latest-live advancement - * cannot drift from transcript ordering. - */ -export function isObserverEventAfter( - candidate: { timestamp: string; seq: number }, - stored: { timestamp: string; seq: number }, -): boolean { - const candidateTime = Date.parse(candidate.timestamp); - const storedTime = Date.parse(stored.timestamp); - if (Number.isFinite(candidateTime) && Number.isFinite(storedTime)) { - if (candidateTime !== storedTime) { - return candidateTime > storedTime; - } - } - return candidate.seq > stored.seq; -} - // Observer event kind for a batch envelope wrapping multiple events. The ACP // harness publishes one frame per second; everything that accumulated between // ticks arrives as `{ kind: "batch", payload: { events: [...] } }` with every @@ -476,9 +456,32 @@ function processLiveObserverEvents( // `control_result` from re-settling a live model switch, and likewise // prevents any other side-effect listener (latest-live tracking, management // requests, session-config capture, lifecycle) from firing twice for one - // frame. Every such listener is a command or idempotent cache write — none - // depends on duplicate re-delivery — so deduping is strictly correct. + // frame. + // + // `accepted` is also the payload external-store subscribers receive; + // `observerChanged` additionally tracks circuit-only changes, which append + // nothing but must still repaint the badge (see the circuit loop below). const accepted = appendAgentEvents(agentPubkey, events); + let observerChanged = accepted !== null; + + // Circuit state is driven by the FULL envelope, not the accepted subset. + // appendAgentEvents' per-agent eviction floor and applyCircuitEvent's + // per-slot ordering gate compare against different reference points, so a + // circuit event can be rejected by one and accepted by the other (e.g. a + // late-delivered frame after a relay reconnect, at/before the agent's floor + // but the first event ever seen for its slot). Feeding this pass only the + // accepted events would silently drop those transitions. Replay is not a + // concern here the way it is for the dispatch loop below: applyCircuitEvent + // has its own per-slot ordering gate and rejects re-arriving frames. + // OR the two signals so a circuit-only change still notifies subscribers, + // matching every other ingestion path in this file + // (ingestArchivedObserverEvents, injectObserverEventsForE2E, + // syncAgentObserverEvents). + for (const parsed of events) { + if (applyCircuitEvent(agentPubkey, parsed)) { + observerChanged = true; + } + } for (const parsed of accepted ?? []) { // Track the latest-live-session-id per (agent, channel) on the live path. @@ -526,6 +529,10 @@ function processLiveObserverEvents( // before specialized callbacks, but external-store subscribers publish once. if (accepted) { notifyListeners({ agentPubkey, events: accepted }); + } else if (observerChanged) { + // Circuit-only change: no events were retained, so there is no targeted + // payload to publish, but circuit subscribers still need waking. + notifyListeners(); } } @@ -842,6 +849,12 @@ export async function ingestArchivedObserverEvents( try { const parsed = (await _decryptFn(event)) as ObserverEvent; for (const inner of unwrapObserverBatch(parsed)) { + // Circuit state is derived independent of channel scope, so apply it + // regardless of which branch below routes the raw event. + if (applyCircuitEvent(agentPubkey, inner)) { + archiveChanged = true; + } + // Route archived events to the channel-scoped archive window (no cap) // rather than the per-agent live-relay store (MAX_OBSERVER_EVENTS cap). // Events without a channelId fall through to the live store so they @@ -884,8 +897,21 @@ export function injectObserverEventsForE2E( events: ObserverEvent[], ) { const added = appendAgentEvents(agentPubkey, events); + // Circuit events must still be folded even when appendAgentEvents rejects + // the raw journal entry (see the matching comment in processLiveObserverEvents) — + // this path exists specifically so E2E specs exercise the real ingestion + // pipeline, and a circuit_open/circuit_recovered event injected here that + // silently failed to update circuit state would make that guarantee false. + let circuitChanged = false; + for (const event of events) { + if (applyCircuitEvent(agentPubkey, event)) circuitChanged = true; + } if (added) { + // The targeted payload also wakes circuit subscribers, so one notify covers + // both; a circuit-only change still needs the untargeted broadcast. notifyListeners({ agentPubkey, events: added }); + } else if (circuitChanged) { + notifyListeners(); } } @@ -898,8 +924,14 @@ export function syncAgentObserverEvents( events: ObserverEvent[], ) { const added = appendAgentEvents(agentPubkey, events); + let circuitChanged = false; + for (const event of events) { + if (applyCircuitEvent(agentPubkey, event)) circuitChanged = true; + } if (added) { notifyListeners({ agentPubkey, events: added }); + } else if (circuitChanged) { + notifyListeners(); } } @@ -913,6 +945,7 @@ export function resetAgentObserverStore() { transcriptByAgent.clear(); evictionFloorByAgent.clear(); snapshotByAgent.clear(); + resetCircuitState(); archiveEventsByChannel.clear(); knownAgentPubkeys.clear(); knownAgentsBySubscription.clear(); diff --git a/desktop/src/features/agents/ui/ManagedAgentRow.tsx b/desktop/src/features/agents/ui/ManagedAgentRow.tsx index 3d205ba3cce..282fe6cff77 100644 --- a/desktop/src/features/agents/ui/ManagedAgentRow.tsx +++ b/desktop/src/features/agents/ui/ManagedAgentRow.tsx @@ -7,8 +7,13 @@ import { PresenceDot } from "@/features/presence/ui/PresenceBadge"; import { Badge } from "@/shared/ui/badge"; import { AgentStatusBadge } from "@/features/agents/ui/AgentStatusBadge"; import { useAgentWorking } from "@/features/agents/agentWorkingSignal"; +import { circuitCooldownRemainingMs } from "@/features/agents/observerRelayStore"; +import { useAgentCircuitStatus } from "@/features/agents/agentCircuitHooks"; import { useOpenAgentActivity } from "@/features/agents/useOpenAgentActivity"; -import { formatElapsed } from "@/features/agents/ui/agentSessionUtils"; +import { + formatDurationMs, + formatElapsed, +} from "@/features/agents/ui/agentSessionUtils"; import { useNow } from "@/shared/lib/useNow"; import type { ManagedAgent, @@ -23,6 +28,7 @@ import { ManagedAgentLogPanel } from "./ManagedAgentLogPanel"; import { PubKey } from "@/shared/ui/PubKey"; import { SubsectionLabel } from "@/shared/ui/PageHeader"; import { resolveModelLabel } from "@/features/agents/lib/formatAgentModelLabel"; +import { Tooltip, TooltipContent, TooltipTrigger } from "@/shared/ui/tooltip"; import { RestartDiffBadge } from "./RestartDiffBadge"; export function ManagedAgentRow({ @@ -60,6 +66,7 @@ export function ManagedAgentRow({ : null; const presenceStatus = presenceLookup[agent.pubkey.trim().toLowerCase()]; const activeTurns = useAgentWorking(agent.pubkey).channels; + const circuitStatus = useAgentCircuitStatus(agent.pubkey); const activeWorkingChannels = React.useMemo( () => activeTurns @@ -120,6 +127,7 @@ export function ManagedAgentRow({ presenceStatus={presenceStatus} /> ; +}) { + const { goChannel } = useAppNavigation(); + const { channelId, message } = circuitStatus; + const activate = channelId + ? () => { + void goChannel(channelId); + } + : undefined; + + // Ticks every second so the cooldown countdown stays live. Shared timer + // across same-interval useNow consumers (see this repo's CLAUDE.md) — cheap + // even with many suspended agents in the list at once. + const now = useNow(1000); + const remainingMs = circuitCooldownRemainingMs(circuitStatus, now); + const label = + remainingMs === null + ? "Suspended — repeated crashes" + : remainingMs > 0 + ? `Suspended — retrying in ${formatDurationMs(remainingMs)}` + : "Suspended — health check pending"; + + return ( + + {/* asChild renders the trigger as the Badge's — see + RestartDiffBadge for why this must stay a non-nested-interactive + element rather than a real @@ -231,46 +259,116 @@ export function BotActivityComposerAction({ side="top" sideOffset={8} > -
- Agents working -
-
- {workingAgents.map((agent) => { - const isSelected = selectedPubkey === agent.pubkey.toLowerCase(); + {hasSuspended ? ( + { + clearHoverTimer(); + setOpen(false); + onOpenAgentSession(pubkey, channelId); + }} + selectedPubkey={selectedPubkey} + trailingIcon={ + + } + variant="destructive" + /> + ) : null} + {workingAgents.length > 0 ? ( + { + clearHoverTimer(); + setOpen(false); + onOpenAgentSession(pubkey, channelId); + }} + selectedPubkey={selectedPubkey} + trailingIcon={ + + } + variant="default" + /> + ) : null} + + + ); +} - return ( - - ); - })} -
- - + )} + data-testid={`bot-activity-composer-item-${agent.pubkey}`} + key={agent.pubkey} + onClick={() => onCloseWithSelection(agent.pubkey)} + type="button" + > + + {agent.name} + + {itemStatusLabel} + + {trailingIcon} + + ); + })} + + ); } diff --git a/desktop/src/features/channels/ui/ChannelComposerActivityAccessory.tsx b/desktop/src/features/channels/ui/ChannelComposerActivityAccessory.tsx index f99888f0112..cdd28ecf64f 100644 --- a/desktop/src/features/channels/ui/ChannelComposerActivityAccessory.tsx +++ b/desktop/src/features/channels/ui/ChannelComposerActivityAccessory.tsx @@ -42,19 +42,21 @@ export function ChannelComposerActivityAccessory({ >
{cardMintJobs.length > 0 ? : null} - {workingBotPubkeys.length > 0 ? ( -
- -
- ) : null} + {/* BotActivityComposerAction also renders for suspended (circuit-open) + agents, not just working ones — its own internal check decides + whether there's anything to show, so this wrapper no longer gates + on workingBotPubkeys alone. */} +
+ +
{typingPubkeys.length > 0 ? ( 0; + // A suspended (circuit-open) agent is never "working", so it wouldn't + // otherwise open this reserved rail — without this, a user typing to an + // agent that just crashed sees no in-channel sign of it at all, only the + // Agents settings screen (see BotActivityComposerAction, which renders the + // matching suspended state once this gate lets it mount). + const composerSuspendedAgents = useOpenCircuitAgents(activityAgents); + const hasComposerBotActivity = + composerWorkingBotPubkeys.length > 0 || composerSuspendedAgents.length > 0; const hasCardMintActivity = useCardMintJobs().length > 0; const hasComposerBottomActivity = hasComposerBotActivity || hasTypingActivity || hasCardMintActivity; diff --git a/desktop/src/features/workflows/ui/WorkflowsView.tsx b/desktop/src/features/workflows/ui/WorkflowsView.tsx index f0dfb1b7dd0..96e5391b619 100644 --- a/desktop/src/features/workflows/ui/WorkflowsView.tsx +++ b/desktop/src/features/workflows/ui/WorkflowsView.tsx @@ -104,17 +104,24 @@ export function WorkflowsView({ const editorWorkflowId = editor && editor.mode !== "create" ? editor.workflowId : null; + // `channels` (from WorkflowsRouteScreen) already includes open channels + // the owner hasn't joined — those are readable without membership, so a + // workflow living there must still surface in this aggregate view rather + // than being silently dropped (see WorkflowsRouteScreen.tsx). Membership + // is still required to create/edit a workflow, so the create-dialog + // channel picker stays scoped to memberChannels below. const memberChannels = channels.filter((c) => c.isMember); - const channelIds = memberChannels.map((c) => c.id).sort(); + const channelIds = channels.map((c) => c.id).sort(); const channelIdKey = channelIds.join(","); const allWorkflowsQuery = useQuery({ queryKey: allWorkflowsQueryKey(channelIdKey), queryFn: async () => { - // Single batched relay query for all member channels, then group by the - // channel_id each workflow carries — replaces the per-channel fanout. + // Single batched relay query for every visible channel (member + open, + // non-member), then group by the channel_id each workflow carries — + // replaces the per-channel fanout. const channelNameById = new Map( - memberChannels.map((channel) => [channel.id, channel.name]), + channels.map((channel) => [channel.id, channel.name]), ); const workflows = await getChannelsWorkflows(channelIds); const results: WorkflowWithChannel[] = []; @@ -128,7 +135,7 @@ export function WorkflowsView({ } return results; }, - enabled: memberChannels.length > 0, + enabled: channelIds.length > 0, ...workflowListFocusRefetchPolicy, }); diff --git a/desktop/src/testing/e2eBridge.ts b/desktop/src/testing/e2eBridge.ts index 6a246572b45..8df3411659b 100644 --- a/desktop/src/testing/e2eBridge.ts +++ b/desktop/src/testing/e2eBridge.ts @@ -2929,6 +2929,36 @@ const mockChannels: MockChannel[] = [ createMockMember(MOCK_IDENTITY_PUBKEY, "member", 540), ], }), + // Private channel the mock identity is NOT a member of (unlike every other + // private fixture above, which all include MOCK_IDENTITY_PUBKEY). Exercises + // the workflow-visibility boundary: an open non-member channel's workflows + // must surface in the aggregate Workflows view, but a private non-member + // channel's must stay hidden. See workflows.spec.ts. + createMockChannel({ + id: "5ec4e700-0000-4000-8000-000000000099", + name: "exec-private", + channel_type: "stream", + visibility: "private", + description: "Private channel the mock identity has not joined", + topic: null, + purpose: null, + last_message_at: null, + archived_at: null, + created_by: ALICE_PUBKEY, + topic_set_by: null, + topic_set_at: null, + purpose_set_by: null, + purpose_set_at: null, + topic_required: false, + max_members: null, + nip29_group_id: null, + created_minutes_ago: 500, + updated_minutes_ago: 500, + members: [ + createMockMember(ALICE_PUBKEY, "owner", 500), + createMockMember(BOB_PUBKEY, "member", 480), + ], + }), createMockChannel({ id: "f48efb06-0c93-5025-aac9-2e646bb6bfa8", name: "alice-tyler", diff --git a/desktop/tests/e2e/circuit-open-badge-screenshots.spec.ts b/desktop/tests/e2e/circuit-open-badge-screenshots.spec.ts new file mode 100644 index 00000000000..304d896124c --- /dev/null +++ b/desktop/tests/e2e/circuit-open-badge-screenshots.spec.ts @@ -0,0 +1,213 @@ +/** + * Screenshot spec for the persistent circuit-breaker "Suspended" badge on the + * real Agents view (block/buzz#5888). + * + * An adversarial review of this feature found the badge had first been wired + * into ManagedAgentRow/AgentGroupRows — a component tree with no reachable + * route in the shipped app — so it had zero real-world effect. It was + * relocated to StandaloneAgentCard/AgentPersonaCard in UnifiedAgentsSection, + * the actual cards this repo's own "Agents" screen renders. This spec proves + * the badge renders there, not just that the underlying store logic is + * correct (the unit tests in observerRelayStore.circuitStatus.test.mjs cover + * that layer already). + * + * Exercises: + * - Badge absent while the circuit is closed. + * - Badge appears on a circuit_open event, with an accessible tooltip + * (Tooltip/TooltipTrigger, not a bare `title` attribute — see the + * RestartDiffBadge precedent this reuses). + * - Badge disappears on a matching circuit_recovered event. + * - Same coverage on the persona-linked card variant. + */ + +import { expect, test } from "@playwright/test"; + +import { installMockBridge, TEST_IDENTITIES } from "../helpers/bridge"; +import { waitForAnimations } from "../helpers/animations"; + +const SHOTS = "test-results/circuit-open-badge-screenshots"; + +const STANDALONE_AGENT = { + pubkey: TEST_IDENTITIES.alice.pubkey, + name: "Local Agent", + status: "running" as const, +}; + +const PERSONA_AGENT = { + pubkey: TEST_IDENTITIES.bob.pubkey, + name: "Persona Agent", + personaId: "builtin:fizz", + status: "running" as const, +}; + +async function waitForSeedHook(page: import("@playwright/test").Page) { + await page.waitForFunction( + () => typeof window.__BUZZ_E2E_SEED_OBSERVER_EVENTS__ === "function", + null, + { timeout: 10_000 }, + ); +} + +async function gotoAgentsView(page: import("@playwright/test").Page) { + await page.goto("/", { waitUntil: "domcontentloaded" }); + await waitForSeedHook(page); + await expect(page.getByTestId("open-agents-view")).toBeVisible({ + timeout: 10_000, + }); + await page.getByTestId("open-agents-view").click(); + await expect(page.getByTestId("agents-library-personas")).toBeVisible({ + timeout: 10_000, + }); +} + +async function seedObserverEvents( + page: import("@playwright/test").Page, + agentPubkey: string, + events: Array<{ + seq: number; + timestamp: string; + kind: string; + agentIndex: number | null; + channelId: string | null; + sessionId: string | null; + turnId: string | null; + payload: unknown; + }>, +) { + await page.evaluate( + ({ pubkey, evts }) => { + window.__BUZZ_E2E_SEED_OBSERVER_EVENTS__?.({ + agentPubkey: pubkey, + events: evts, + }); + }, + { pubkey: agentPubkey, evts: events }, + ); + // Let React re-render after the store update. + await page.waitForTimeout(300); +} + +// No cooldown_secs in the payload — keeps the badge on its static +// "Suspended — repeated crashes" label rather than a live countdown, so the +// screenshot and text assertions aren't timing-sensitive. +function circuitOpenEvent(overrides: { channelId?: string | null } = {}) { + return { + seq: 1, + timestamp: new Date().toISOString(), + kind: "circuit_open", + agentIndex: 0, + channelId: overrides.channelId ?? null, + sessionId: null, + turnId: null, + payload: { + error: + "Agent slot 0 panicked repeatedly and its circuit breaker is now open.", + }, + }; +} + +function circuitRecoveredEvent() { + return { + seq: 2, + timestamp: new Date(Date.now() + 1000).toISOString(), + kind: "circuit_recovered", + agentIndex: 0, + channelId: null, + sessionId: null, + turnId: null, + payload: { error: "Agent slot 0 recovered." }, + }; +} + +test.describe("circuit-open badge screenshots", () => { + test.use({ viewport: { width: 1280, height: 900 } }); + + test.beforeEach(async ({ page }) => { + page.on("pageerror", (err) => { + console.error( + "PAGE ERROR:", + err.message, + err.stack?.split("\n").slice(0, 5).join("\n"), + ); + }); + }); + + test("01-standalone-card-circuit-open-and-recover", async ({ page }) => { + await installMockBridge(page, { managedAgents: [STANDALONE_AGENT] }); + await gotoAgentsView(page); + + const agentCard = page.getByTestId( + `managed-agent-${STANDALONE_AGENT.pubkey}`, + ); + await expect(agentCard).toBeVisible({ timeout: 10_000 }); + const badge = agentCard.getByTestId( + `managed-agent-circuit-open-${STANDALONE_AGENT.pubkey}`, + ); + + // Closed by default — no badge. + await expect(badge).toHaveCount(0); + + await seedObserverEvents(page, STANDALONE_AGENT.pubkey, [ + circuitOpenEvent(), + ]); + + await expect(badge).toBeVisible({ timeout: 5_000 }); + await expect(badge).toHaveText("Suspended — repeated crashes"); + + // Accessible tooltip: keyboard-focusable (tabIndex), Tooltip primitive + // rather than a bare `title` attribute. + await badge.hover(); + const tooltip = page.locator("[role=tooltip]"); + await expect(tooltip).toBeVisible({ timeout: 5_000 }); + await expect(tooltip).toHaveText( + "Agent slot 0 panicked repeatedly and its circuit breaker is now open.", + ); + + await waitForAnimations(page); + await agentCard.screenshot({ + path: `${SHOTS}/01-standalone-card-circuit-open.png`, + }); + + await seedObserverEvents(page, STANDALONE_AGENT.pubkey, [ + circuitRecoveredEvent(), + ]); + + await expect(badge).toHaveCount(0); + }); + + test("02-persona-card-circuit-open-and-recover", async ({ page }) => { + await installMockBridge(page, { + activePersonaIds: [PERSONA_AGENT.personaId], + managedAgents: [PERSONA_AGENT], + }); + await gotoAgentsView(page); + + const personaCard = page.getByTestId( + `persona-agent-row-${PERSONA_AGENT.personaId}`, + ); + await expect(personaCard).toBeVisible({ timeout: 10_000 }); + const badge = personaCard.getByTestId( + `managed-agent-circuit-open-${PERSONA_AGENT.pubkey}`, + ); + + await expect(badge).toHaveCount(0); + + await seedObserverEvents(page, PERSONA_AGENT.pubkey, [ + circuitOpenEvent({ channelId: "94a444a4-c0a3-5966-ab05-530c6ddc2301" }), + ]); + + await expect(badge).toBeVisible({ timeout: 5_000 }); + await expect(badge).toHaveText("Suspended — repeated crashes"); + + await waitForAnimations(page); + await personaCard.screenshot({ + path: `${SHOTS}/02-persona-card-circuit-open.png`, + }); + + await seedObserverEvents(page, PERSONA_AGENT.pubkey, [ + circuitRecoveredEvent(), + ]); + + await expect(badge).toHaveCount(0); + }); +}); diff --git a/desktop/tests/e2e/observer-feed-screenshots.spec.ts b/desktop/tests/e2e/observer-feed-screenshots.spec.ts index 44ff609c9ee..393ad512d27 100644 --- a/desktop/tests/e2e/observer-feed-screenshots.spec.ts +++ b/desktop/tests/e2e/observer-feed-screenshots.spec.ts @@ -814,4 +814,65 @@ test.describe("observer feed screenshots", () => { path: `${SHOTS}/11-first-turn-ordering.png`, }); }); + + test("12 — circuit_open and circuit_recovered lifecycle alerts", async ({ + page, + }) => { + await installMockBridge(page, { managedAgents: MANAGED_AGENTS }); + const feedPanel = await openObserverFeedPanel(page, OBSERVER_AGENT_PUBKEY); + + // Circuit breaker opened after a panic inside this channel (the real + // panicked-in-channel call site passes the triggering channel's id, see + // emit_circuit_open_alert's caller in buzz-acp/src/lib.rs), then + // recovered. Both events are seeded with this channel's id to prove the + // renderer works when a channel context is present. The real + // circuit_recovered alert now carries that same real channel_id too — + // it's threaded through from the SlotCircuit that opened it (see + // emit_circuit_recovered_alert's caller in buzz-acp/src/lib.rs) rather + // than always being channel_id=None. Both the transcript bubble here and + // the persistent ManagedAgentRow badge + // (data-testid="managed-agent-circuit-open") reflect it. + await seedObserverEvents(page, OBSERVER_AGENT_PUBKEY, [ + { + seq: 1, + timestamp: NOW, + kind: "circuit_open", + agentIndex: 0, + channelId: CHANNEL_ID, + sessionId: null, + turnId: null, + payload: { + trigger: "panicked", + cooldown_secs: 300, + error: + "Agent slot 0 panicked repeatedly and its circuit breaker is now open " + + "— it will not respond until the 300s cooldown elapses and a health probe succeeds.", + }, + }, + { + seq: 2, + timestamp: NOW, + kind: "circuit_recovered", + agentIndex: 0, + channelId: CHANNEL_ID, + sessionId: null, + turnId: null, + payload: { + error: + "Agent slot 0 recovered — its circuit breaker probe succeeded and it is responding again.", + }, + }, + ]); + + await expect( + feedPanel.getByText("Agent suspended (repeated crashes)"), + ).toBeVisible({ timeout: 5_000 }); + await expect(feedPanel.getByText("Agent recovered")).toBeVisible({ + timeout: 5_000, + }); + await settleAnimations(feedPanel); + await feedPanel.screenshot({ + path: `${SHOTS}/12-circuit-breaker-alerts.png`, + }); + }); }); diff --git a/desktop/tests/e2e/workflows.spec.ts b/desktop/tests/e2e/workflows.spec.ts index 2e26f57382e..3b35930506e 100644 --- a/desktop/tests/e2e/workflows.spec.ts +++ b/desktop/tests/e2e/workflows.spec.ts @@ -7,6 +7,50 @@ test.beforeEach(async ({ page }) => { await installMockBridge(page); }); +// Fixture channel ids from src/testing/e2eBridge.ts. "sales" is an OPEN +// channel the mock identity has not joined; "exec-private" is a PRIVATE +// channel the mock identity has not joined. Together they pin the +// visibility boundary WorkflowsRouteScreen/WorkflowsView enforce: open +// non-member channels surface in the aggregate view, private non-member +// channels stay hidden. +const SALES_OPEN_NON_MEMBER_CHANNEL_ID = "c6f3a9b2-4d55-5a23-bf78-5b9e2g3c5d6f"; +const EXEC_PRIVATE_NON_MEMBER_CHANNEL_ID = + "5ec4e700-0000-4000-8000-000000000099"; + +async function seedWorkflowInChannel( + page: import("@playwright/test").Page, + channelId: string, + name: string, +) { + const yamlDefinition = [ + `name: ${name}`, + "enabled: true", + "trigger:", + " on: manual", + "steps:", + " - name: step_1", + ].join("\n"); + + await page.evaluate( + async ({ channelId: id, yamlDefinition: yaml }) => { + const testWindow = window as Window & { + __BUZZ_E2E_INVOKE_MOCK_COMMAND__?: ( + command: string, + payload?: Record, + ) => Promise; + }; + if (!testWindow.__BUZZ_E2E_INVOKE_MOCK_COMMAND__) { + throw new Error("__BUZZ_E2E_INVOKE_MOCK_COMMAND__ is not installed"); + } + await testWindow.__BUZZ_E2E_INVOKE_MOCK_COMMAND__("create_workflow", { + channelId: id, + yamlDefinition: yaml, + }); + }, + { channelId, yamlDefinition }, + ); +} + async function navigateToWorkflows(page: import("@playwright/test").Page) { await page.goto("/"); await page.getByTestId("open-workflows-view").click(); @@ -1008,3 +1052,51 @@ test("missing workflow routes show an unavailable modal with close and retry", a await unavailable.getByRole("button", { name: "Close" }).click(); await expect(page).toHaveURL(/#\/workflows$/); }); + +test("surfaces a workflow from an open channel the owner has not joined", async ({ + page, +}) => { + const workflowName = `sales_open_nonmember_${Date.now()}`; + await page.goto("/"); + await page.getByTestId("app-sidebar").waitFor({ state: "visible" }); + await seedWorkflowInChannel( + page, + SALES_OPEN_NON_MEMBER_CHANNEL_ID, + workflowName, + ); + + await page.getByTestId("open-workflows-view").click(); + await expect(page).toHaveURL(/#\/workflows$/); + await expect(page.getByTestId("workflows-view")).toBeVisible(); + + const card = page + .locator('[data-testid^="workflow-card-"]') + .filter({ hasText: workflowName }); + await expect(card).toBeVisible(); + await waitForAnimations(page); + await card.screenshot({ + path: "test-results/screenshots/workflow-visible-open-nonmember-channel.png", + }); +}); + +test("keeps a workflow from a private channel the owner has not joined out of the aggregate view", async ({ + page, +}) => { + const workflowName = `private_nonmember_${Date.now()}`; + await page.goto("/"); + await page.getByTestId("app-sidebar").waitFor({ state: "visible" }); + await seedWorkflowInChannel( + page, + EXEC_PRIVATE_NON_MEMBER_CHANNEL_ID, + workflowName, + ); + + await page.getByTestId("open-workflows-view").click(); + await expect(page).toHaveURL(/#\/workflows$/); + await expect(page.getByTestId("workflows-view")).toBeVisible(); + + await expect(page.getByText("No workflows yet")).toBeVisible(); + await expect( + page.getByTestId("workflows-view").getByText(workflowName), + ).toHaveCount(0); +});