Show active thread count for agents

Signed-off-by: Salman Mohammed <smohammed@squareup.com>

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<observer::ObserverHandle>,
     agent_index: Option<usize>,
     mut context: observer::ObserverContext,
+    thread_root_event_id: Option<String>,
     interval: Duration,
     state: Arc<Mutex<LivenessState>>,
 ) {
@@ -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,
         );

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, [

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<string, number>();
+  const earliestByChannel = new Map<
+    string,
+    { startedAt: number; threadRoots: Set<string>; 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);

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/);
+});

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<ManagedAgent, "pubkey" | "name">;
+export type BotActivityAgent = Pick<ManagedAgent, "pubkey" | "name"> &
+  Partial<Pick<ManagedAgent, "sessionPolicy">>;
 
 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 (

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"
             />

desktop/src/features/channels/ui/useChannelAgentSessions.ts

@@ -20,6 +20,7 @@ import {
 import type { PanelValueSetter } from "./useChannelPanelHistoryState";
 
 export type ChannelAgentSessionAgent = Pick<ManagedAgent, "pubkey" | "name"> & {
+  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,