feat(desktop): observer-driven activity bar in the thread composer

desktop/src/features/agents/activeAgentTurnsStore.test.mjs

@@ -12,6 +12,8 @@ import {
   restoreActiveAgentTurnsForCommunity,
   clearSavedCommunitySnapshot,
   clearActiveTurnsForAgent,
+  getWorkingAgentPubkeysForThread,
+  getActiveTurnCountForChannel,
 } from "./activeAgentTurnsStore.ts";
 import {
   injectObserverEventsForE2E,
@@ -2018,3 +2020,110 @@ describe("clearActiveTurnsForAgent", () => {
     mock.timers.reset();
   });
 });
+
+describe("getWorkingAgentPubkeysForThread", () => {
+  beforeEach(() => {
+    resetActiveAgentTurnsStore();
+  });
+
+  it("matches a turn whose sessionId prefixes the thread root id", () => {
+    syncAgentTurnsFromEvents(AGENT, [
+      makeEvent({
+        sessionId: "aaaa000011112222",
+        turnId: "aaaa000011112222-1700000000000",
+      }),
+    ]);
+
+    assert.deepEqual(
+      [
+        ...getWorkingAgentPubkeysForThread(
+          "chan-1",
+          "aaaa000011112222deadbeefdeadbeef",
+        ),
+      ],
+      [AGENT],
+    );
+    assert.equal(
+      getWorkingAgentPubkeysForThread(
+        "chan-1",
+        "bbbb000011112222deadbeefdeadbeef",
+      ).length,
+      0,
+    );
+    assert.equal(
+      getWorkingAgentPubkeysForThread(
+        "chan-2",
+        "aaaa000011112222deadbeefdeadbeef",
+      ).length,
+      0,
+    );
+  });
+
+  it("ignores turns without a sessionId", () => {
+    syncAgentTurnsFromEvents(AGENT, [makeEvent({ sessionId: null })]);
+
+    assert.equal(
+      getWorkingAgentPubkeysForThread("chan-1", "any-root-id").length,
+      0,
+    );
+  });
+
+  it("drops the agent once its turn completes", () => {
+    syncAgentTurnsFromEvents(AGENT, [
+      makeEvent({ sessionId: "aaaa000011112222" }),
+    ]);
+    syncAgentTurnsFromEvents(AGENT, [
+      makeEvent({
+        seq: 2,
+        kind: "turn_completed",
+        timestamp: "2024-01-01T00:01:00Z",
+      }),
+    ]);
+
+    assert.equal(
+      getWorkingAgentPubkeysForThread(
+        "chan-1",
+        "aaaa000011112222deadbeefdeadbeef",
+      ).length,
+      0,
+    );
+  });
+});
+
+describe("getActiveTurnCountForChannel", () => {
+  beforeEach(() => {
+    resetActiveAgentTurnsStore();
+  });
+
+  it("counts only this agent's live turns in this channel", () => {
+    syncAgentTurnsFromEvents(AGENT, [
+      makeEvent({ seq: 1, turnId: "t-a", sessionId: "aaaa" }),
+      makeEvent({ seq: 2, turnId: "t-b", sessionId: "bbbb" }),
+      makeEvent({ seq: 3, turnId: "t-c", channelId: "chan-2" }),
+    ]);
+    syncAgentTurnsFromEvents(AGENT_2, [
+      makeEvent({ seq: 1, turnId: "t-d", sessionId: "dddd" }),
+    ]);
+
+    assert.equal(getActiveTurnCountForChannel(AGENT, "chan-1"), 2);
+    assert.equal(getActiveTurnCountForChannel(AGENT, "chan-2"), 1);
+    assert.equal(getActiveTurnCountForChannel(AGENT_2, "chan-1"), 1);
+  });
+
+  it("returns 0 for missing inputs and completed turns", () => {
+    assert.equal(getActiveTurnCountForChannel(null, "chan-1"), 0);
+    assert.equal(getActiveTurnCountForChannel(AGENT, null), 0);
+
+    syncAgentTurnsFromEvents(AGENT, [makeEvent({ seq: 1, turnId: "t-a" })]);
+    syncAgentTurnsFromEvents(AGENT, [
+      makeEvent({
+        seq: 2,
+        turnId: "t-a",
+        kind: "turn_completed",
+        timestamp: "2024-01-01T00:01:00Z",
+      }),
+    ]);
+
+    assert.equal(getActiveTurnCountForChannel(AGENT, "chan-1"), 0);
+  });
+});

desktop/src/features/agents/activeAgentTurnsStore.ts

@@ -40,6 +40,10 @@ const PRUNE_INTERVAL_MS = 5_000;
 type ActiveTurn = {
   turnId: string;
   channelId: string;
+  /** Thread-root prefix the emitting harness scopes this turn to (NIP-AO
+   *  envelope `sessionId`, the root event id shortened). Null when the frame
+   *  carried none — such turns are channel-scoped only. */
+  sessionId: string | null;
   startedAt: number;
   lastActivityAt: number;
 };
@@ -84,6 +88,9 @@ const clockOffsetByAgent = new Map<string, number>();
 // Only regenerated when the underlying turn map for an agent actually changes.
 const cachedTurnSummaries = new Map<string, ActiveTurnSummary[]>();
 let cachedChannelTurnSummaries: ActiveChannelTurnSummary[] | null = null;
+// `${channelId}|${threadRootId}` → working pubkeys, reference-stable per notify.
+const cachedThreadWorkingPubkeys = new Map<string, string[]>();
+const EMPTY_THREAD_PUBKEYS: readonly string[] = Object.freeze([]);
 
 // Composite watermark per agent: the newest observer event processed, by
 // (timestamp, seq) ordering. An event is processed only if it is strictly
@@ -104,6 +111,8 @@ let pruneInterval: ReturnType<typeof setInterval> | null = null;
 function invalidateCache(agentKey: string) {
   cachedTurnSummaries.delete(agentKey);
   cachedChannelTurnSummaries = null;
+  // Thread-scoped snapshots span agents, so any agent change clears them all.
+  cachedThreadWorkingPubkeys.clear();
 }
 
 function notifyListeners() {
@@ -139,6 +148,7 @@ function startTurn(
   channelId: string,
   turnId: string,
   timestamp: string,
+  sessionId: string | null = null,
 ) {
   const key = normalizePubkey(agentPubkey);
   let agentTurns = activeTurnsByAgent.get(key);
@@ -166,6 +176,7 @@ function startTurn(
   agentTurns.set(turnId, {
     turnId,
     channelId,
+    sessionId,
     startedAt,
     lastActivityAt: Date.now(),
   });
@@ -214,7 +225,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,
+    event.sessionId ?? null,
+  );
   return true;
 }
 
@@ -353,6 +370,7 @@ function processEvent(agentPubkey: string, event: ObserverEvent) {
           event.channelId,
           event.turnId ?? `seq-${event.seq}`,
           event.timestamp,
+          event.sessionId ?? null,
         );
         notifyListeners();
         return;
@@ -546,6 +564,72 @@ export function useActiveAgentTurnsByChannel(): ActiveChannelTurnSummary[] {
   );
 }
 
+/**
+ * Live observer turns one agent is running in one channel. The channel
+ * composer bar uses this to switch from a detailed single-turn status line to
+ * an aggregate ("2 threads") when parallel conversations would otherwise
+ * interleave in one line. Returns a primitive, so useSyncExternalStore
+ * reference stability is free.
+ */
+export function getActiveTurnCountForChannel(
+  agentPubkey: string | null | undefined,
+  channelId: string | null,
+): number {
+  if (!agentPubkey || !channelId) return 0;
+  const turns = activeTurnsByAgent.get(normalizePubkey(agentPubkey));
+  if (!turns) return 0;
+  let count = 0;
+  for (const turn of turns.values()) {
+    if (turn.channelId === channelId) count += 1;
+  }
+  return count;
+}
+
+/**
+ * Agents with a live observer turn scoped to one thread. A turn qualifies
+ * when its channel matches AND the thread root id begins with the turn's
+ * `sessionId` (harnesses emit the root event id shortened, so prefix match is
+ * the identity test). Turns without a sessionId never qualify — the channel
+ * bar covers those.
+ */
+export function getWorkingAgentPubkeysForThread(
+  channelId: string | null,
+  threadRootId: string | null,
+): readonly string[] {
+  if (!channelId || !threadRootId) return EMPTY_THREAD_PUBKEYS;
+  const cacheKey = `${channelId}|${threadRootId}`;
+  const cached = cachedThreadWorkingPubkeys.get(cacheKey);
+  if (cached) return cached;
+
+  const pubkeys: string[] = [];
+  for (const [agentKey, turns] of activeTurnsByAgent) {
+    for (const turn of turns.values()) {
+      if (turn.channelId !== channelId) continue;
+      if (!turn.sessionId || !threadRootId.startsWith(turn.sessionId)) continue;
+      pubkeys.push(agentKey);
+      break;
+    }
+  }
+  cachedThreadWorkingPubkeys.set(cacheKey, pubkeys);
+  return pubkeys;
+}
+
+/**
+ * Hook: agents working in one specific thread, observer-derived.
+ * Reference-stable between store notifications.
+ */
+export function useThreadWorkingAgentPubkeys(
+  channelId: string | null,
+  threadRootId: string | null,
+): readonly string[] {
+  const getSnapshot = React.useCallback(
+    () => getWorkingAgentPubkeysForThread(channelId, threadRootId),
+    [channelId, threadRootId],
+  );
+
+  return React.useSyncExternalStore(subscribeActiveAgentTurns, getSnapshot);
+}
+
 /**
  * Sync every running/deployed agent's observer events into the active-turns
  * store. Extracted from the bridge hook so a regression can drive the exact