From 5335e120dd16d3ace48c001caa4abd2444ab1dc1 Mon Sep 17 00:00:00 2001 From: Salman Mohammed Date: Fri, 25 Sep 2026 15:38:08 -0400 Subject: [PATCH] Show active thread count for agents Signed-off-by: Salman Mohammed --- crates/buzz-acp/src/pool.rs | 44 +++--- .../agents/activeAgentTurnsStore.test.mjs | 44 ++++++ .../features/agents/activeAgentTurnsStore.ts | 63 ++++++++- .../channels/ui/BotActivityBar.jsdom-test.mjs | 131 ++++++++++++++++++ .../features/channels/ui/BotActivityBar.tsx | 31 ++++- .../ui/ChannelComposerActivityAccessory.tsx | 1 + .../channels/ui/useChannelAgentSessions.ts | 2 + 7 files changed, 287 insertions(+), 29 deletions(-) create mode 100644 desktop/src/features/channels/ui/BotActivityBar.jsdom-test.mjs diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 15df4f63f8..c434503830 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -2335,6 +2335,7 @@ pub async fn run_prompt_task( None => PromptSource::Heartbeat, }; let observer_channel_id = source.channel_id(); + let observer_thread_root = source.scope().and_then(SessionScope::root_event_id); let turn_started_at = chrono::Utc::now().to_rfc3339(); agent.acp.set_observer_context(observer::context_for_turn( observer_channel_id, @@ -2346,16 +2347,17 @@ pub async fn run_prompt_task( .as_ref() .map(|b| b.events.iter().map(|be| be.event.id.to_hex()).collect()) .unwrap_or_default(); - agent.acp.observe( - "turn_started", - serde_json::json!({ - "source": match &source { - PromptSource::Channel(_) => "channel", - PromptSource::Heartbeat => "heartbeat", - }, - "triggeringEventIds": triggering_event_ids, - }), - ); + let mut turn_started_payload = serde_json::json!({ + "source": match &source { + PromptSource::Channel(_) => "channel", + PromptSource::Heartbeat => "heartbeat", + }, + "triggeringEventIds": triggering_event_ids, + }); + if let Some(root) = observer_thread_root { + turn_started_payload["threadRootEventId"] = serde_json::json!(root); + } + agent.acp.observe("turn_started", turn_started_payload); // Emits `turn_completed` on any exit path. Captures observer handle and // metadata now, before the agent is moved into PromptResult. It must be @@ -2390,6 +2392,7 @@ pub async fn run_prompt_task( turn_id.clone(), turn_started_at.clone(), ), + observer_thread_root.map(str::to_owned), ctx.turn_liveness_interval, Arc::clone(&liveness_state), ); @@ -5016,6 +5019,7 @@ async fn run_turn_liveness( observer: Option, agent_index: Option, mut context: observer::ObserverContext, + thread_root_event_id: Option, interval: Duration, state: Arc>, ) { @@ -5043,12 +5047,11 @@ async fn run_turn_liveness( return; } context.session_id = guard.session_id.clone(); - observer.emit( - "turn_liveness", - agent_index, - &context, - serde_json::json!({}), - ); + let payload = match &thread_root_event_id { + Some(root) => serde_json::json!({ "threadRootEventId": root }), + None => serde_json::json!({}), + }; + observer.emit("turn_liveness", agent_index, &context, payload); drop(guard); } } @@ -9172,6 +9175,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" Some(observer.clone()), Some(0), context, + None, Duration::from_secs(10), Arc::clone(&state), )), @@ -9222,6 +9226,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" Some(observer.clone()), Some(0), context, + None, Duration::from_secs(10), Arc::clone(&state), )), @@ -9274,6 +9279,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" Some(observer.clone()), Some(0), context, + Some("root-1".into()), Duration::from_secs(10), Arc::clone(&state), )), @@ -9297,6 +9303,9 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" .filter(|e| e.kind == "turn_liveness") .collect(); assert_eq!(pings.len(), 2); + assert!(pings + .iter() + .all(|ping| ping.payload["threadRootEventId"] == "root-1")); assert_eq!( pings[0].session_id, None, "pre-resolution ping must not carry a session ID" @@ -9316,6 +9325,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" Some(observer.clone()), Some(0), context, + None, Duration::ZERO, open_liveness_state(), ); @@ -9339,6 +9349,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" None, None, context, + None, Duration::from_secs(10), open_liveness_state(), ); @@ -9378,6 +9389,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" Some(observer.clone()), Some(0), context, + None, Duration::from_secs(10), state, ); diff --git a/desktop/src/features/agents/activeAgentTurnsStore.test.mjs b/desktop/src/features/agents/activeAgentTurnsStore.test.mjs index 2658469b91..3801bf81ac 100644 --- a/desktop/src/features/agents/activeAgentTurnsStore.test.mjs +++ b/desktop/src/features/agents/activeAgentTurnsStore.test.mjs @@ -54,6 +54,50 @@ describe("activeAgentTurnsStore", () => { resetActiveAgentTurnsStore(); }); + it("counts distinct active thread roots in a channel and drops completed roots", () => { + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ + seq: 1, + turnId: "t1", + payload: { threadRootEventId: "ROOT-A" }, + }), + makeEvent({ + seq: 2, + turnId: "t2", + payload: { threadRootEventId: "root-a" }, + }), + makeEvent({ + seq: 3, + turnId: "t3", + payload: { threadRootEventId: "root-b" }, + }), + ]); + assert.equal(getActiveTurnsForAgent(AGENT)[0].threadCount, 2); + + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ seq: 4, kind: "turn_completed", turnId: "t3" }), + ]); + assert.equal(getActiveTurnsForAgent(AGENT)[0].threadCount, 1); + }); + + it("learns a turn's thread root from a liveness event", () => { + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ seq: 1, turnId: "t1", payload: null }), + makeEvent({ + seq: 2, + kind: "turn_liveness", + turnId: "t1", + payload: { threadRootEventId: "root-a" }, + }), + makeEvent({ + seq: 3, + turnId: "t2", + payload: { threadRootEventId: "root-a" }, + }), + ]); + assert.equal(getActiveTurnsForAgent(AGENT)[0].threadCount, 1); + }); + describe("seq filtering", () => { it("processes events with increasing seq", () => { syncAgentTurnsFromEvents(AGENT, [ diff --git a/desktop/src/features/agents/activeAgentTurnsStore.ts b/desktop/src/features/agents/activeAgentTurnsStore.ts index ebafb3f197..0ba859efbb 100644 --- a/desktop/src/features/agents/activeAgentTurnsStore.ts +++ b/desktop/src/features/agents/activeAgentTurnsStore.ts @@ -54,6 +54,7 @@ const PRUNE_INTERVAL_MS = 5_000; type ActiveTurn = { turnId: string; channelId: string; + threadRootEventId: string | null; startedAt: number; lastActivityAt: number; }; @@ -62,6 +63,10 @@ type ActiveTurn = { export type ActiveTurnSummary = { channelId: string; anchorAt: number; + /** Distinct active thread roots; legacy frames use their turn ID. */ + threadCount?: number; + /** True when the observer identifies at least one thread-scoped turn. */ + hasThreadScope?: boolean; }; /** One channel with active agent work, aggregated across agents. */ @@ -174,11 +179,21 @@ function parseTimestamp(timestamp: string): number | null { return Number.isFinite(parsed) ? parsed : null; } +function threadRootFromEvent(event: ObserverEvent): string | null { + if (!event.payload || typeof event.payload !== "object") return null; + const root = (event.payload as { threadRootEventId?: unknown }) + .threadRootEventId; + return typeof root === "string" && root.length > 0 + ? root.toLowerCase() + : null; +} + function startTurn( agentPubkey: string, channelId: string, turnId: string, timestamp: string, + threadRootEventId: string | null = null, ) { const key = normalizePubkey(agentPubkey); let agentTurns = activeTurnsByAgent.get(key); @@ -206,13 +221,18 @@ function startTurn( agentTurns.set(turnId, { turnId, channelId, + threadRootEventId, startedAt, lastActivityAt: Date.now(), }); invalidateCache(key); } -function recordActivity(agentPubkey: string, turnId: string | null): boolean { +function recordActivity( + agentPubkey: string, + turnId: string | null, + threadRootEventId: string | null, +): boolean { if (!turnId) return false; const key = normalizePubkey(agentPubkey); const agentTurns = activeTurnsByAgent.get(key); @@ -220,6 +240,11 @@ function recordActivity(agentPubkey: string, turnId: string | null): boolean { const turn = agentTurns.get(turnId); if (turn) { turn.lastActivityAt = Date.now(); + if (threadRootEventId && turn.threadRootEventId !== threadRootEventId) { + turn.threadRootEventId = threadRootEventId; + invalidateCache(key); + notifyListeners(); + } return true; } return false; @@ -254,7 +279,13 @@ function resurrectTurn(agentPubkey: string, event: ObserverEvent): boolean { frameAt !== null && startedAtMs !== null && startedAtMs <= frameAt ? startedAt : event.timestamp; - startTurn(agentPubkey, event.channelId, event.turnId, safeStartedAt); + startTurn( + agentPubkey, + event.channelId, + event.turnId, + safeStartedAt, + threadRootFromEvent(event), + ); return true; } @@ -408,6 +439,7 @@ function processEvent(agentPubkey: string, event: ObserverEvent) { event.channelId, event.turnId ?? `seq-${event.seq}`, event.timestamp, + threadRootFromEvent(event), ); notifyListeners(); return; @@ -432,7 +464,11 @@ function processEvent(agentPubkey: string, event: ObserverEvent) { // turn was pruned out from under a still-running host (a transient drop // raced the pause, or the lone-crash residual self-healed), resurrect it. case "turn_liveness": { - const refreshed = recordActivity(agentPubkey, event.turnId ?? null); + const refreshed = recordActivity( + agentPubkey, + event.turnId ?? null, + threadRootFromEvent(event), + ); if (!refreshed && resurrectTurn(agentPubkey, event)) { notifyListeners(); return; @@ -508,18 +544,31 @@ export function getActiveTurnsForAgent( // Collapse multiple turns in one channel to the earliest start — the badge // should count from when the channel's oldest live turn began. Anchors are // derived here (startedAt + offset) so the latest skew estimate applies. - const earliestByChannel = new Map(); + const earliestByChannel = new Map< + string, + { startedAt: number; threadRoots: Set; hasThreadScope: boolean } + >(); for (const turn of agentTurns.values()) { const prior = earliestByChannel.get(turn.channelId); - if (prior === undefined || turn.startedAt < prior) { - earliestByChannel.set(turn.channelId, turn.startedAt); + if (!prior) { + earliestByChannel.set(turn.channelId, { + startedAt: turn.startedAt, + threadRoots: new Set([turn.threadRootEventId ?? turn.turnId]), + hasThreadScope: turn.threadRootEventId !== null, + }); + } else { + prior.startedAt = Math.min(prior.startedAt, turn.startedAt); + prior.threadRoots.add(turn.threadRootEventId ?? turn.turnId); + prior.hasThreadScope ||= turn.threadRootEventId !== null; } } const result = [...earliestByChannel.entries()] - .map(([channelId, startedAt]) => ({ + .map(([channelId, { startedAt, threadRoots, hasThreadScope }]) => ({ channelId, anchorAt: startedAt + offset, + threadCount: threadRoots.size, + hasThreadScope, })) .sort((a, b) => a.channelId.localeCompare(b.channelId)); cachedTurnSummaries.set(key, result); diff --git a/desktop/src/features/channels/ui/BotActivityBar.jsdom-test.mjs b/desktop/src/features/channels/ui/BotActivityBar.jsdom-test.mjs new file mode 100644 index 0000000000..c9af85c6a5 --- /dev/null +++ b/desktop/src/features/channels/ui/BotActivityBar.jsdom-test.mjs @@ -0,0 +1,131 @@ +import assert from "node:assert/strict"; +import { registerHooks } from "node:module"; +import { afterEach, test } from "node:test"; + +import React, { act } from "react"; +import { createRoot } from "react-dom/client"; + +registerHooks({ + resolve(specifier, context, nextResolve) { + if (specifier === "@/features/agents/ui/useObserverEvents") { + return { shortCircuit: true, url: "buzz-test-stub:agent-transcript" }; + } + return nextResolve(specifier, context); + }, + load(url, context, nextLoad) { + if (url === "buzz-test-stub:agent-transcript") { + return { + format: "module", + shortCircuit: true, + source: "export function useAgentTranscript() { return []; }\n", + }; + } + return nextLoad(url, context); + }, +}); + +const { BotActivityComposerAction } = await import("./BotActivityBar.tsx"); +const { resetActiveAgentTurnsStore, syncAgentTurnsFromEvents } = await import( + "@/features/agents/activeAgentTurnsStore" +); + +const AGENT = "a".repeat(64); +let mounted = null; + +afterEach(async () => { + if (mounted) { + await act(async () => mounted.root.unmount()); + mounted.container.remove(); + mounted = null; + } + resetActiveAgentTurnsStore(); +}); + +function turnEvent(seq, turnId, kind, root) { + return { + seq, + timestamp: new Date(Date.now() + seq).toISOString(), + kind, + agentIndex: 0, + channelId: "channel-1", + sessionId: "session-1", + turnId, + payload: { threadRootEventId: root }, + }; +} + +async function mountBar(sessionPolicy, showChannelThreadCount = true) { + const container = document.createElement("div"); + document.body.appendChild(container); + const root = createRoot(container); + mounted = { container, root }; + await act(async () => { + root.render( + React.createElement(BotActivityComposerAction, { + agents: [{ pubkey: AGENT, name: "Caddie", sessionPolicy }], + channelId: "channel-1", + onOpenAgentSession: () => {}, + openAgentSessionPubkey: null, + showChannelThreadCount, + workingBotPubkeys: [AGENT], + variant: "inline", + }), + ); + }); + return container; +} + +test("thread-policy composer shows thread count only while multiple roots are active", async () => { + resetActiveAgentTurnsStore(); + syncAgentTurnsFromEvents(AGENT, [ + turnEvent(1, "turn-a", "turn_started", "root-a"), + ]); + const container = await mountBar("thread"); + assert.match(container.textContent, /Caddie: Working/); + + await act(async () => { + syncAgentTurnsFromEvents(AGENT, [ + turnEvent(2, "turn-b", "turn_started", "root-b"), + ]); + }); + assert.match(container.textContent, /Caddie is working on 2 threads now/); + + await act(async () => { + syncAgentTurnsFromEvents(AGENT, [ + turnEvent(3, "turn-b", "turn_completed", "root-b"), + ]); + }); + assert.match(container.textContent, /Caddie: Working/); +}); + +test("channel-policy composer keeps its current label with multiple turns", async () => { + resetActiveAgentTurnsStore(); + syncAgentTurnsFromEvents(AGENT, [ + turnEvent(1, "turn-a", "turn_started", null), + turnEvent(2, "turn-b", "turn_started", null), + ]); + const container = await mountBar("channel"); + assert.match(container.textContent, /Caddie: Working/); + assert.doesNotMatch(container.textContent, /threads now/); +}); + +test("observer thread roots work when the agent has no local policy metadata", async () => { + resetActiveAgentTurnsStore(); + syncAgentTurnsFromEvents(AGENT, [ + turnEvent(1, "turn-a", "turn_started", "root-a"), + turnEvent(2, "turn-b", "turn_started", "root-b"), + ]); + const container = await mountBar(undefined); + assert.match(container.textContent, /Caddie is working on 2 threads now/); +}); + +test("thread reply composer does not show a channel-wide thread count", async () => { + resetActiveAgentTurnsStore(); + syncAgentTurnsFromEvents(AGENT, [ + turnEvent(1, "turn-a", "turn_started", "root-a"), + turnEvent(2, "turn-b", "turn_started", "root-b"), + ]); + const container = await mountBar("thread", false); + assert.match(container.textContent, /Caddie: Working/); + assert.doesNotMatch(container.textContent, /threads now/); +}); diff --git a/desktop/src/features/channels/ui/BotActivityBar.tsx b/desktop/src/features/channels/ui/BotActivityBar.tsx index dacdc7e89e..e73c08e4bb 100644 --- a/desktop/src/features/channels/ui/BotActivityBar.tsx +++ b/desktop/src/features/channels/ui/BotActivityBar.tsx @@ -1,6 +1,7 @@ import * as React from "react"; import { Loader2 } from "lucide-react"; +import { useActiveAgentTurns } from "@/features/agents/activeAgentTurnsStore"; import { useAgentTranscript } from "@/features/agents/ui/useObserverEvents"; import { getActivityHeadline, @@ -19,7 +20,8 @@ import { import { Shimmer } from "@/shared/ui/Shimmer"; import { UserAvatar } from "@/shared/ui/UserAvatar"; -export type BotActivityAgent = Pick; +export type BotActivityAgent = Pick & + Partial>; type BotActivityBarProps = { agents: BotActivityAgent[]; @@ -27,6 +29,7 @@ type BotActivityBarProps = { onOpenAgentSession: (pubkey: string, channelId?: string | null) => void; openAgentSessionPubkey: string | null; profiles?: UserProfileLookup; + showChannelThreadCount?: boolean; workingBotPubkeys: string[]; variant?: "toolbar" | "inline"; }; @@ -40,6 +43,7 @@ export function BotActivityComposerAction({ onOpenAgentSession, openAgentSessionPubkey, profiles, + showChannelThreadCount = false, workingBotPubkeys, variant = "toolbar", }: BotActivityBarProps) { @@ -57,6 +61,17 @@ export function BotActivityComposerAction({ }, [agents, workingBotPubkeys]); const singleWorkingAgent = workingAgents.length === 1 ? (workingAgents[0] ?? null) : null; + const activeTurns = useActiveAgentTurns( + showChannelThreadCount ? singleWorkingAgent?.pubkey : null, + ); + const activeChannelTurns = activeTurns.find( + (turn) => turn.channelId === channelId, + ); + const activeThreadCount = + singleWorkingAgent?.sessionPolicy === "thread" || + activeChannelTurns?.hasThreadScope + ? (activeChannelTurns?.threadCount ?? 0) + : 0; const transcript = useAgentTranscript( Boolean(singleWorkingAgent), singleWorkingAgent?.pubkey, @@ -149,15 +164,19 @@ export function BotActivityComposerAction({ const selectedPubkey = openAgentSessionPubkey?.toLowerCase() ?? null; const triggerLabel = workingAgents.length === 1 - ? `${workingAgents[0]?.name ?? "Agent"} is working` + ? activeThreadCount > 1 + ? `${workingAgents[0]?.name ?? "Agent"} is working on ${activeThreadCount} threads now` + : `${workingAgents[0]?.name ?? "Agent"} is working` : `${workingAgents.length} agents working`; const isInline = variant === "inline"; const visibleStatusLabel = workingAgents.length === 1 - ? `${workingAgents[0]?.name ?? "Agent"}: ${ - activityHeadlines[headlineIndex % activityHeadlines.length] ?? - "Working" - }` + ? activeThreadCount > 1 + ? `${workingAgents[0]?.name ?? "Agent"} is working on ${activeThreadCount} threads now` + : `${workingAgents[0]?.name ?? "Agent"}: ${ + activityHeadlines[headlineIndex % activityHeadlines.length] ?? + "Working" + }` : `${workingAgents[0]?.name ?? "Agent"} +${workingAgents.length - 1}`; return ( diff --git a/desktop/src/features/channels/ui/ChannelComposerActivityAccessory.tsx b/desktop/src/features/channels/ui/ChannelComposerActivityAccessory.tsx index f99888f011..298f82b526 100644 --- a/desktop/src/features/channels/ui/ChannelComposerActivityAccessory.tsx +++ b/desktop/src/features/channels/ui/ChannelComposerActivityAccessory.tsx @@ -50,6 +50,7 @@ export function ChannelComposerActivityAccessory({ onOpenAgentSession={onOpenAgentSession} openAgentSessionPubkey={openAgentSessionPubkey} profiles={profiles} + showChannelThreadCount workingBotPubkeys={workingBotPubkeys} variant="inline" /> diff --git a/desktop/src/features/channels/ui/useChannelAgentSessions.ts b/desktop/src/features/channels/ui/useChannelAgentSessions.ts index d5cd1824e0..90c95d0375 100644 --- a/desktop/src/features/channels/ui/useChannelAgentSessions.ts +++ b/desktop/src/features/channels/ui/useChannelAgentSessions.ts @@ -20,6 +20,7 @@ import { import type { PanelValueSetter } from "./useChannelPanelHistoryState"; export type ChannelAgentSessionAgent = Pick & { + sessionPolicy?: ManagedAgent["sessionPolicy"]; status: ManagedAgent["status"] | "unknown"; agentSource: "managed" | "member-bot" | "relay"; canInterruptTurn: boolean; @@ -86,6 +87,7 @@ export function buildChannelAgentSessionCandidates({ name: agent.name, status: agent.status, agentSource: "managed", + sessionPolicy: agent.sessionPolicy, canInterruptTurn: true, channelIds: existing?.channelIds, channels: existing?.channels,