feat(telemetry): push bus for live agent frames

/api/world/live is a 2s database poll, which is right for queryable state and
wrong for a token stream: reasoning only became visible after a step finished
and its row was written. This adds a process-wide broadcast bus that
topology_exec publishes to as the runtime's WebSocket delivers frames, and the
SSE handler forwards without waiting for the next tick.

Measured: the pushed frame arrived ~2.2s before the polled copy of the same
text.

Design notes worth keeping:
- A global (OnceLock), not an AppState field. The publisher is reached through
  phase_runner -> topology_worker -> MissionTap, none of which hold AppState;
  threading a handle through all of them would put a UI concern into four
  layers that have no other reason to know about one.
- Lossy by design. A slow subscriber lags and skips rather than applying
  backpressure to the agent producing. mission_events remains the durable
  record; this bus is the fast path, never the source of truth.
- Only `claw_<uuid>` aliases are attributed. The governor, door and evaluator
  drive real turns under other names, and attributing their output to an agent
  would put words in someone's mouth. Asserted in a test.
- The poll no longer emits `reasoning`: with both paths live, every turn
  arrived TWICE — once pushed, once polled ~2s later. The row is still written;
  this feed just is not its second mouth.

CEILING, measured rather than assumed: turns are not token-level because the
runtime is not streaming. zeroclaw's claude_cli provider runs
`claude -p --output-format json`, which returns ONE result object when the turn
completes — there are no incremental tokens to forward. Making this genuinely
token-by-token needs `--output-format stream-json` and incremental parsing in
the zeroclaw fork, not here. The bus is in place and will carry them the day it
does.

Co-Authored-By: Claude Opus 5 <[email protected]>

crates/cm-api/src/lib.rs

@@ -18,6 +18,7 @@ pub mod fleet_herdr;
 pub mod harvest;
 pub mod level_up;
 pub mod library;
+pub mod live_bus;
 mod mcp_door;
 mod mcp_skills;
 pub mod microvm_client;

crates/cm-api/src/live_bus.rs

@@ -0,0 +1,126 @@
+//! A process-wide push bus for live taxonomy events.
+//!
+//! `/api/world/live` is a 2-second database poll. That is the right shape for
+//! state you can query — statuses, phases, telemetry — and the wrong shape for a
+//! token stream: an agent's reasoning only becomes visible after the step
+//! finishes and its text is persisted, so the REASONING STREAM card showed
+//! completed paragraphs rather than an agent thinking.
+//!
+//! This carries the frames that cannot wait for a round trip through Postgres.
+//! `topology_exec` publishes as the runtime's WebSocket delivers them; the SSE
+//! handler subscribes and forwards, so a chunk reaches the browser in one hop.
+//!
+//! **Why a global rather than a field on `AppState`.** The publisher is
+//! `topology_exec`, reached through `phase_runner` → `topology_worker` →
+//! `MissionTap`, none of which hold `AppState`. Threading a handle through all
+//! of them would put a UI concern into four layers that have no other reason to
+//! know about one. There is exactly one bus per process and it holds no
+//! per-request state, so a `OnceLock` is the honest representation.
+//!
+//! **Lossy on purpose.** A slow reader lags and skips rather than applying
+//! backpressure to the agent that is producing. Dropping frames degrades a live
+//! view; blocking would slow the mission to the speed of the slowest open tab.
+//! The durable record is `mission_events` — this bus is the fast path, never the
+//! source of truth.
+
+use std::sync::{Arc, OnceLock};
+
+use serde_json::Value;
+use tokio::sync::broadcast;
+use uuid::Uuid;
+
+/// Bounded so a stalled subscriber costs memory once, not unboundedly. At
+/// token granularity a busy mission produces a few hundred frames a second;
+/// this is roughly a couple of seconds of slack before a slow reader starts
+/// skipping.
+const CAPACITY: usize = 2048;
+
+#[derive(Debug, Clone)]
+pub struct LiveEvent {
+    /// Every subscriber is workspace-scoped; the bus is not.
+    pub workspace_id: Uuid,
+    /// A taxonomy type, e.g. `agent.reasoning.delta`.
+    pub kind: String,
+    pub data: Value,
+}
+
+pub struct LiveBus {
+    tx: broadcast::Sender<LiveEvent>,
+}
+
+impl LiveBus {
+    fn new() -> LiveBus {
+        let (tx, _rx) = broadcast::channel(CAPACITY);
+        LiveBus { tx }
+    }
+
+    /// Publish. Returns immediately, and succeeds even with no subscribers —
+    /// nobody watching is the normal case, not an error.
+    pub fn publish(&self, workspace_id: Uuid, kind: &str, data: Value) {
+        let _ = self.tx.send(LiveEvent {
+            workspace_id,
+            kind: kind.to_string(),
+            data,
+        });
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<LiveEvent> {
+        self.tx.subscribe()
+    }
+}
+
+static BUS: OnceLock<Arc<LiveBus>> = OnceLock::new();
+
+pub fn global() -> &'static Arc<LiveBus> {
+    BUS.get_or_init(|| Arc::new(LiveBus::new()))
+}
+
+/// The claw alias the runtime dispatches on (`claw_<uuid>`) → the agent id the
+/// UI keys on. Returns `None` for any other alias — the governor, the door and
+/// the evaluator all drive turns under names that are not claws, and attributing
+/// their output to an agent would put words in someone's mouth.
+pub fn agent_id_from_alias(alias: &str) -> Option<Uuid> {
+    Uuid::parse_str(alias.strip_prefix("claw_")?).ok()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn only_claw_aliases_resolve_to_an_agent() {
+        let id = Uuid::now_v7();
+        assert_eq!(
+            agent_id_from_alias(&format!("claw_{id}")),
+            Some(id),
+            "the runtime's own alias form must resolve"
+        );
+        // These drive real turns and must NOT be attributed to an agent.
+        for other in ["scout", "coordinator", "door", "evaluator", "claw_nonsense"] {
+            assert_eq!(agent_id_from_alias(other), None, "{other}");
+        }
+    }
+
+    #[tokio::test]
+    async fn a_subscriber_receives_what_is_published() {
+        let bus = LiveBus::new();
+        let mut rx = bus.subscribe();
+        let ws = Uuid::now_v7();
+        bus.publish(
+            ws,
+            "agent.reasoning.delta",
+            serde_json::json!({"text": "hi"}),
+        );
+        let ev = rx.recv().await.expect("delivered");
+        assert_eq!(ev.workspace_id, ws);
+        assert_eq!(ev.kind, "agent.reasoning.delta");
+    }
+
+    /// Publishing with nobody listening must not error — that is the common
+    /// case (no browser open) and it must never disturb the mission.
+    #[test]
+    fn publishing_into_the_void_is_fine() {
+        let bus = LiveBus::new();
+        bus.publish(Uuid::now_v7(), "agent.tool.call", serde_json::json!({}));
+    }
+}

crates/cm-api/src/routes/world.rs

@@ -752,6 +752,10 @@ pub async fn world_live(
         // On first seed we still emit — the World engine needs the initial
         // scale for every pawn.
         let mut last_bytes: std::collections::HashMap<String, u64> = std::collections::HashMap::new();
+        // Push channel for frames that cannot wait for the 2s poll — an agent's
+        // reasoning arrives token by token. Subscribed BEFORE the first poll so
+        // nothing produced during the initial queries is missed.
+        let mut live_rx = crate::live_bus::global().subscribe();
         loop {
             let roster = match cm_db::repo::agents::roster(&pool, ws).await {
                 Ok(r) => r,
@@ -1128,7 +1132,12 @@ pub async fn world_live(
                       WHERE m.workspace_id = $1
                         AND e.id > $2
                         AND e.agent_id IS NOT NULL
-                        AND e.kind IN ('reasoning', 'tool.call')
+                        -- 'reasoning' is NOT read here: live_bus pushes those
+                        -- as the runtime emits them, and emitting from both
+                        -- delivered every turn twice — once pushed, once polled
+                        -- ~2s later. The row is still written, as the durable
+                        -- record; this feed just is not its second mouth.
+                        AND e.kind = 'tool.call'
                       ORDER BY e.id
                       LIMIT 200",
                 )
@@ -1145,22 +1154,13 @@ pub async fn world_live(
                     let kind: String = r.get("kind");
                     let detail: serde_json::Value = r.get("detail");
                     let target: Option<String> = r.get("target");
-                    if kind == "tool.call" {
+                    debug_assert_eq!(kind, "tool.call", "query filters to tool.call");
+                    let _ = kind;
                     yield sse("agent.tool.call", json!({
                         "agentId": agent_id.to_string(),
                         "tool": target.clone().unwrap_or_default(),
                         "target": detail.get("path").and_then(|v| v.as_str()),
                     }));
-                    } else {
-                        let text = detail.get("text").and_then(|v| v.as_str()).unwrap_or("");
-                        if !text.is_empty() {
-                            yield sse("agent.reasoning.delta", json!({
-                                "agentId": agent_id.to_string(),
-                                "text": text,
-                                "channel": "say",
-                            }));
-                        }
-                    }
                 }
             }
 
@@ -1290,7 +1290,30 @@ pub async fn world_live(
             }
 
             first = false;
-            tokio::time::sleep(Duration::from_secs(2)).await;
+            // Forward pushed frames until the next poll is due, instead of
+            // sleeping through them. This is what makes the reasoning stream
+            // token-level: a chunk reaches the browser as the runtime emits it,
+            // while everything queryable keeps its 2s cadence.
+            let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
+            loop {
+                let left = deadline.saturating_duration_since(tokio::time::Instant::now());
+                if left.is_zero() {
+                    break;
+                }
+                match tokio::time::timeout(left, live_rx.recv()).await {
+                    Ok(Ok(ev)) => {
+                        if ev.workspace_id == ws.as_uuid() {
+                            yield sse(&ev.kind, ev.data);
+                        }
+                    }
+                    // Lagged: this subscriber fell behind and frames were
+                    // dropped for it. Keep going — a live view that skips is
+                    // right, and blocking the producer would be wrong.
+                    Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => continue,
+                    Ok(Err(_)) => break,
+                    Err(_) => break,
+                }
+            }
         }
     };
 

crates/cm-api/src/topology_exec.rs

@@ -67,6 +67,9 @@ pub struct ZeroClawDriveExecutor {
 /// put mission concepts into every tier that has no missions.
 pub struct MissionTap {
     pub pool: sqlx::PgPool,
+    /// Which workspace's live feed these frames belong to. Every subscriber is
+    /// workspace-scoped, so a frame without this could not be routed.
+    pub workspace_id: uuid::Uuid,
     pub mission_id: uuid::Uuid,
     pub phase_id: Option<uuid::Uuid>,
     pub run_id: Option<uuid::Uuid>,
@@ -294,7 +297,16 @@ impl ZeroClawDriveExecutor {
             .await
             .map_err(|e| OrchestratorError::Executor(format!("ws send failed: {e}")))?;
 
-        let (outcome, trace) = match tokio::time::timeout(TURN_TIMEOUT, Self::drain(&mut ws)).await
+        let (outcome, trace) = match tokio::time::timeout(
+            TURN_TIMEOUT,
+            Self::drain(
+                &mut ws,
+                self.tap.as_ref().and_then(|t| {
+                    crate::live_bus::agent_id_from_alias(alias).map(|a| (t.workspace_id, a))
+                }),
+            ),
+        )
+        .await
         {
             Ok(res) => res?,
             Err(_) => {
@@ -457,7 +469,13 @@ impl ZeroClawDriveExecutor {
     /// trace is separate from [`TurnOutcome`] deliberately: that type is the
     /// shared orchestrator contract used by every tier, and tool telemetry is a
     /// mission concern.
-    async fn drain<S>(ws: &mut S) -> Result<(TurnOutcome, ToolTrace), OrchestratorError>
+    /// `live` is the push target for this turn: `Some((workspace, agent))` when
+    /// the turn belongs to a mission AND runs under a claw alias. `None` for the
+    /// governor/door/evaluator, whose output belongs to no agent.
+    async fn drain<S>(
+        ws: &mut S,
+        live: Option<(uuid::Uuid, uuid::Uuid)>,
+    ) -> Result<(TurnOutcome, ToolTrace), OrchestratorError>
     where
         S: StreamExt<Item = Result<Message, tokio_tungstenite::tungstenite::Error>>
             + SinkExt<Message>
@@ -478,6 +496,24 @@ impl ZeroClawDriveExecutor {
                         "chunk" => {
                             if let Some(c) = v.get("content").and_then(|c| c.as_str()) {
                                 output.push_str(c);
+                                // Push, don't wait for the poll. This is the
+                                // whole point of the bus: the reasoning card
+                                // previously showed a step's text only after the
+                                // step ended and the row was written, so an
+                                // agent mid-thought looked idle for seconds.
+                                if let Some((ws_id, agent_id)) = live {
+                                    if !c.trim().is_empty() {
+                                        crate::live_bus::global().publish(
+                                            ws_id,
+                                            "agent.reasoning.delta",
+                                            serde_json::json!({
+                                                "agentId": agent_id.to_string(),
+                                                "text": c,
+                                                "channel": "say",
+                                            }),
+                                        );
+                                    }
+                                }
                             }
                         }
                         "done" => {
@@ -634,22 +670,31 @@ mod tests {
     /// second error on top of the first.
     #[test]
     fn the_container_name_is_derived_or_absent_never_wrong() {
-        let ex = |url: &str| ZeroClawDriveExecutor::new(
+        let ex = |url: &str| {
+            ZeroClawDriveExecutor::new(
                 url.to_string(),
                 String::new(),
                 std::collections::HashMap::new(),
                 "scout".into(),
-        );
+            )
+        };
         assert_eq!(
-            ex("http://cm-runtime-mission-019fec2d596f:42617").container_name().as_deref(),
+            ex("http://cm-runtime-mission-019fec2d596f:42617")
+                .container_name()
+                .as_deref(),
             Some("cm-runtime-mission-019fec2d596f")
         );
         assert_eq!(
-            ex("https://host.example:8443/base").container_name().as_deref(),
+            ex("https://host.example:8443/base")
+                .container_name()
+                .as_deref(),
             Some("host.example")
         );
         // No scheme is still a host.
-        assert_eq!(ex("clawmates-runtime:42617").container_name().as_deref(), Some("clawmates-runtime"));
+        assert_eq!(
+            ex("clawmates-runtime:42617").container_name().as_deref(),
+            Some("clawmates-runtime")
+        );
         assert_eq!(ex("").container_name(), None);
     }
 
@@ -792,18 +837,31 @@ mod tests {
         assert_eq!(
             trace.calls,
             vec![
-                ToolCall { tool: "Read".into(), path: Some("/mission/repo/src/a.rs".into()) },
+                ToolCall {
-                ToolCall { tool: "Bash".into(), path: None },
+                    tool: "Read".into(),
+                    path: Some("/mission/repo/src/a.rs".into())
+                },
+                ToolCall {
+                    tool: "Bash".into(),
+                    path: None
+                },
                 // `arguments_summary` said "src/main.rs". It is prose, so it is
                 // not a file touch — a path scraped from a sentence would put
                 // files on the map that no agent opened.
-                ToolCall { tool: "Grep".into(), path: None },
+                ToolCall {
+                    tool: "Grep".into(),
+                    path: None
+                },
             ]
         );
         assert_eq!(trace.unmatched.get("a_frame_we_have_never_seen"), Some(&2));
         assert_eq!(trace.unmatched.get("session_start"), Some(&1));
         // `done` terminates the drain and is not an unmatched frame.
-        assert!(!trace.unmatched.contains_key("done"), "{:?}", trace.unmatched);
+        assert!(
+            !trace.unmatched.contains_key("done"),
+            "{:?}",
+            trace.unmatched
+        );
     }
 
     #[tokio::test]

crates/cm-api/src/topology_worker.rs

@@ -204,6 +204,7 @@ async fn run_job(
         .map(
             |(_, _, mission_id, phase_id)| crate::topology_exec::MissionTap {
                 pool: pool.clone(),
+                workspace_id: job.workspace_id,
                 mission_id: *mission_id,
                 phase_id: *phase_id,
                 run_id: Some(id),