agent ui: live event panel via SSE + stream-json

CLAUDE.md

@@ -182,6 +182,24 @@ and the MCP tools. Claude drives any further `recv`/`send` itself —
 harness no longer relays claude's stdout as a reply. Stdout is logged for
 debugging; the side effects (sends via MCP) are what matter.
 
+**Live view.** Each agent runs a `hive_ag3nt::events::Bus` (a
+`tokio::sync::broadcast<LiveEvent>` wrapper). The harness emits:
+- `TurnStart { from, body }` when a wake-up message is popped.
+- `Stream(value)` for every line claude prints on stdout (parsed
+  stream-json; flattened under `{kind: "stream", type: ...}` via serde
+  internal tagging).
+- `Note(text)` for stderr lines and non-JSON stdout (so nothing's lost).
+- `TurnEnd { ok, note }` when claude exits.
+
+The web UI subscribes via `/events/stream` (SSE) and a small JS panel on
+`/` appends rows. No full-page reload — the login form (and anything else
+the operator is typing into) stays put.
+
+claude is invoked with `--print --verbose --output-format stream-json` so
+tool calls + assistant text + tool results all land as structured events.
+The harness no longer reads claude's text stdout into a reply; claude
+calls `mcp__hyperhive__send` itself.
+
 **Tool envelope.** Every MCP tool handler in `hive_ag3nt::mcp::AgentServer`
 wraps its logic in `run_tool(name, args_debug, async { ... })`. The
 envelope guarantees:

Cargo.lock

@@ -457,6 +457,7 @@ dependencies = [
  "serde",
  "serde_json",
  "tokio",
+ "tokio-stream",
  "tracing",
  "tracing-subscriber",
 ]

hive-ag3nt/Cargo.toml

@@ -16,6 +16,7 @@ schemars.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 tokio.workspace = true
+tokio-stream.workspace = true
 tracing.workspace = true
 tracing-subscriber.workspace = true
 

hive-ag3nt/src/bin/hive-ag3nt.rs

@@ -1,12 +1,15 @@
 use std::path::{Path, PathBuf};
+use std::process::Stdio;
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use anyhow::{Result, bail};
 use clap::{Parser, Subcommand};
+use hive_ag3nt::events::{Bus, LiveEvent};
 use hive_ag3nt::login::{self, LoginState};
 use hive_ag3nt::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, mcp, web_ui};
 use hive_sh4re::{AgentRequest, AgentResponse};
+use tokio::io::{AsyncBufReadExt, BufReader};
 use tokio::process::Command;
 
 #[derive(Parser)]
@@ -61,14 +64,16 @@ async fn main() -> Result<()> {
             tracing::info!(state = ?initial, claude_dir = %claude_dir.display(), "harness boot");
             let login_state = Arc::new(Mutex::new(initial));
             let ui_state = login_state.clone();
+            let bus = Bus::new();
+            let ui_bus = bus.clone();
             tokio::spawn(async move {
-                if let Err(e) = web_ui::serve(label, port, ui_state).await {
+                if let Err(e) = web_ui::serve(label, port, ui_state, ui_bus).await {
                     tracing::error!(error = ?e, "web ui failed");
                 }
             });
             match initial {
                 LoginState::Online => {
-                    serve(&cli.socket, Duration::from_millis(poll_ms), login_state).await
+                    serve(&cli.socket, Duration::from_millis(poll_ms), login_state, bus).await
                 }
                 LoginState::NeedsLogin => {
                     // Partial-run mode: keep the harness alive (so the web UI
@@ -77,7 +82,7 @@ async fn main() -> Result<()> {
                     // from the dashboard PTY path in step 4 or via
                     // `root-login` + `claude auth login` in the meantime)
                     // transitions us into the turn loop without a restart.
-                    needs_login_loop(&cli.socket, &claude_dir, login_state, poll_ms).await
+                    needs_login_loop(&cli.socket, &claude_dir, login_state, poll_ms, bus).await
                 }
             }
         }
@@ -103,6 +108,7 @@ async fn needs_login_loop(
     claude_dir: &Path,
     state: Arc<Mutex<LoginState>>,
     poll_ms: u64,
+    bus: Bus,
 ) -> Result<()> {
     tracing::warn!(
         claude_dir = %claude_dir.display(),
@@ -114,12 +120,17 @@ async fn needs_login_loop(
         if login::has_session(claude_dir) {
             tracing::info!("claude session detected — entering turn loop");
             *state.lock().unwrap() = LoginState::Online;
-            return serve(socket, Duration::from_millis(poll_ms), state).await;
+            return serve(socket, Duration::from_millis(poll_ms), state, bus).await;
         }
     }
 }
 
-async fn serve(socket: &Path, interval: Duration, state: Arc<Mutex<LoginState>>) -> Result<()> {
+async fn serve(
+    socket: &Path,
+    interval: Duration,
+    state: Arc<Mutex<LoginState>>,
+    bus: Bus,
+) -> Result<()> {
     tracing::info!(socket = %socket.display(), "hive-ag3nt serve");
     let _ = state; // reserved for future state transitions (turn-loop -> needs-login)
     let mcp_config = write_mcp_config(socket).await?;
@@ -129,10 +140,28 @@ async fn serve(socket: &Path, interval: Duration, state: Arc<Mutex<LoginState>>)
         match recv {
             Ok(AgentResponse::Message { from, body }) => {
                 tracing::info!(%from, %body, "inbox");
+                bus.emit(LiveEvent::TurnStart {
+                    from: from.clone(),
+                    body: body.clone(),
+                });
                 let prompt = format_wake_prompt(&label, &from, &body);
-                match invoke_claude(&prompt, &mcp_config).await {
-                    Ok(out) => tracing::info!(stdout = %out.trim(), "claude turn finished"),
-                    Err(e) => tracing::warn!(error = %format!("{e:#}"), "claude turn failed"),
+                let outcome = run_turn(&prompt, &mcp_config, &bus).await;
+                match outcome {
+                    Ok(()) => {
+                        bus.emit(LiveEvent::TurnEnd {
+                            ok: true,
+                            note: None,
+                        });
+                        tracing::info!("claude turn finished");
+                    }
+                    Err(e) => {
+                        let note = format!("{e:#}");
+                        bus.emit(LiveEvent::TurnEnd {
+                            ok: false,
+                            note: Some(note.clone()),
+                        });
+                        tracing::warn!(error = %note, "claude turn failed");
+                    }
                 }
             }
             Ok(AgentResponse::Empty) => {}
@@ -173,15 +202,20 @@ fn format_wake_prompt(label: &str, from: &str, body: &str) -> String {
     )
 }
 
-async fn invoke_claude(prompt: &str, mcp_config: &Path) -> Result<String> {
-    // Whitelist model: `--tools` restricts which built-ins exist in the
-    // session (omitting WebFetch/WebSearch/Task means claude literally
-    // can't invoke them); `--allowedTools` auto-approves the same set
-    // plus the hyperhive MCP surface so there's no permission prompt
-    // mid-turn. A finer-grained allow-list system for Bash command
-    // patterns is on the backlog (PLAN.md polish).
-    let out = Command::new("claude")
+/// Spawn `claude` for one turn and stream its `stream-json` stdout into
+/// the live event bus. `--verbose` is required by claude-code when pairing
+/// `--print` with `--output-format stream-json`. Each stdout line is one
+/// JSON event; we broadcast the parsed value (or a `Note` fallback on
+/// parse error so the UI doesn't silently lose information). The tool
+/// whitelist is the same as before: omit WebFetch/WebSearch/Task; allow
+/// the hyperhive MCP surface auto-approved. Bash pattern allow-list is on
+/// the backlog (CLAUDE.md).
+async fn run_turn(prompt: &str, mcp_config: &Path, bus: &Bus) -> Result<()> {
+    let mut child = Command::new("claude")
         .arg("--print")
+        .arg("--verbose")
+        .arg("--output-format")
+        .arg("stream-json")
         .arg("--model")
         .arg("haiku")
         .arg("--mcp-config")
@@ -191,20 +225,38 @@ async fn invoke_claude(prompt: &str, mcp_config: &Path) -> Result<String> {
         .arg("--allowedTools")
         .arg(mcp::allowed_tools_arg())
         .arg(prompt)
-        .output()
-        .await?;
-    if !out.status.success() {
-        bail!(
-            "claude exited {}: {}",
-            out.status,
-            String::from_utf8_lossy(&out.stderr).trim()
-        );
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()?;
+
+    let stdout = child.stdout.take().expect("piped stdout");
+    let stderr = child.stderr.take().expect("piped stderr");
+
+    let bus_out = bus.clone();
+    let bus_err = bus.clone();
+    let pump_stdout = tokio::spawn(async move {
+        let mut reader = BufReader::new(stdout).lines();
+        while let Ok(Some(line)) = reader.next_line().await {
+            match serde_json::from_str::<serde_json::Value>(&line) {
+                Ok(v) => bus_out.emit(LiveEvent::Stream(v)),
+                Err(_) => bus_out.emit(LiveEvent::Note(format!("(non-json) {line}"))),
             }
-    let text = String::from_utf8_lossy(&out.stdout).trim().to_owned();
-    if text.is_empty() {
-        bail!("claude produced empty output");
         }
-    Ok(text)
+    });
+    let pump_stderr = tokio::spawn(async move {
+        let mut reader = BufReader::new(stderr).lines();
+        while let Ok(Some(line)) = reader.next_line().await {
+            bus_err.emit(LiveEvent::Note(format!("stderr: {line}")));
+        }
+    });
+
+    let status = child.wait().await?;
+    let _ = pump_stdout.await;
+    let _ = pump_stderr.await;
+    if !status.success() {
+        bail!("claude exited {status}");
+    }
+    Ok(())
 }
 
 /// Drop the per-agent MCP config on disk so the turn loop can hand its path

hive-ag3nt/src/bin/hive-m1nd.rs

@@ -9,6 +9,7 @@ use std::time::Duration;
 
 use anyhow::{Result, bail};
 use clap::{Parser, Subcommand};
+use hive_ag3nt::events::Bus;
 use hive_ag3nt::login::{self, LoginState};
 use hive_ag3nt::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, web_ui};
 use hive_sh4re::{HelperEvent, ManagerRequest, ManagerResponse, SYSTEM_SENDER};
@@ -66,11 +67,14 @@ async fn main() -> Result<()> {
             tracing::info!(state = ?initial, claude_dir = %claude_dir.display(), "hm1nd boot");
             let login_state = Arc::new(Mutex::new(initial));
             let ui_state = login_state.clone();
+            let bus = Bus::new();
+            let ui_bus = bus.clone();
             tokio::spawn(async move {
-                if let Err(e) = web_ui::serve(label, port, ui_state).await {
+                if let Err(e) = web_ui::serve(label, port, ui_state, ui_bus).await {
                     tracing::error!(error = ?e, "web ui failed");
                 }
             });
+            let _ = bus; // manager turn loop not wired to events yet
             match initial {
                 LoginState::Online => serve(&cli.socket, Duration::from_millis(poll_ms)).await,
                 LoginState::NeedsLogin => {

hive-ag3nt/src/events.rs

@@ -0,0 +1,63 @@
+//! Live event stream for the per-agent web UI. The harness emits one
+//! `LiveEvent` per interesting thing that happens during a turn — wake-up
+//! (the popped inbox message), every line claude prints on stdout
+//! (parsed from `--output-format stream-json`), and the turn-end summary.
+//! The web UI subscribes via SSE and renders rows live.
+//!
+//! Channel type is `tokio::sync::broadcast`. New subscribers see only
+//! future events; the dashboard JS deals with the cold-start case by
+//! showing "connecting…" until the first event arrives.
+
+use std::sync::Arc;
+
+use serde::{Deserialize, Serialize};
+use tokio::sync::broadcast;
+
+const CHANNEL_CAPACITY: usize = 256;
+
+/// One row of the agent's live stream. Serialised to JSON for SSE delivery.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+pub enum LiveEvent {
+    /// Harness popped a wake-up message and is about to invoke claude.
+    TurnStart { from: String, body: String },
+    /// One line of claude's `--output-format stream-json` stdout, parsed as
+    /// a generic JSON value (so we don't have to track every claude-code
+    /// event variant). The frontend pretty-prints by `type` field.
+    Stream(serde_json::Value),
+    /// Free-form note from the harness (e.g. "claude exited 0",
+    /// "stream-json parse error: ..."). Useful when stream-json itself
+    /// fails so the UI doesn't just go silent.
+    Note(String),
+    /// Turn finished. `ok=false` means claude exited non-zero or the
+    /// harness hit a transport error.
+    TurnEnd { ok: bool, note: Option<String> },
+}
+
+#[derive(Clone)]
+pub struct Bus {
+    tx: Arc<broadcast::Sender<LiveEvent>>,
+}
+
+impl Bus {
+    #[must_use]
+    pub fn new() -> Self {
+        let (tx, _) = broadcast::channel(CHANNEL_CAPACITY);
+        Self { tx: Arc::new(tx) }
+    }
+
+    pub fn emit(&self, event: LiveEvent) {
+        // Lagged subscribers drop events — fine; the UI is a tail, not a log.
+        let _ = self.tx.send(event);
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<LiveEvent> {
+        self.tx.subscribe()
+    }
+}
+
+impl Default for Bus {
+    fn default() -> Self {
+        Self::new()
+    }
+}

hive-ag3nt/src/lib.rs

@@ -2,6 +2,7 @@
 //! `hive-m1nd` (manager) binaries.
 
 pub mod client;
+pub mod events;
 pub mod login;
 pub mod login_session;
 pub mod mcp;

hive-ag3nt/src/web_ui.rs

@@ -3,6 +3,7 @@
 //! each instance must bind a distinct port. `HIVE_PORT` is set per agent by
 //! `hive-c0re`'s generated per-agent flake (deterministic from agent name).
 
+use std::convert::Infallible;
 use std::net::SocketAddr;
 use std::sync::{Arc, Mutex};
 
@@ -10,11 +11,16 @@ use anyhow::{Context, Result};
 use axum::{
     Form, Router,
     extract::State,
-    response::{Html, IntoResponse, Redirect, Response},
+    response::{
+        Html, IntoResponse, Redirect, Response,
+        sse::{Event, KeepAlive, Sse},
+    },
     routing::{get, post},
 };
 use serde::Deserialize;
+use tokio_stream::{Stream, StreamExt, wrappers::BroadcastStream};
 
+use crate::events::Bus;
 use crate::login::LoginState;
 use crate::login_session::{LoginSession, drop_if_finished};
 
@@ -28,16 +34,19 @@ struct AppState {
     label: String,
     login: LoginStateCell,
     session: Arc<Mutex<Option<Arc<LoginSession>>>>,
+    bus: Bus,
 }
 
-pub async fn serve(label: String, port: u16, login: LoginStateCell) -> Result<()> {
+pub async fn serve(label: String, port: u16, login: LoginStateCell, bus: Bus) -> Result<()> {
     let state = AppState {
         label,
         login,
         session: Arc::new(Mutex::new(None)),
+        bus,
     };
     let app = Router::new()
         .route("/", get(index))
+        .route("/events/stream", get(events_stream))
         .route("/login/start", post(post_login_start))
         .route("/login/code", post(post_login_code))
         .route("/login/cancel", post(post_login_cancel))
@@ -67,9 +76,87 @@ async fn index(State(state): State<AppState>) -> Html<String> {
 }
 
 fn render_online() -> String {
-    "<p class=\"status-online\">▓█▓▒░ harness alive — turn loop running ▓█▓▒░</p>\n<p class=\"meta\">phase 6a placeholder — turn-loop status / inbox / xterm.js coming in 6b+</p>".into()
+    format!(
+        "<p class=\"status-online\">▓█▓▒░ harness alive — turn loop running ▓█▓▒░</p>\n{LIVE_PANEL}",
+    )
 }
 
+/// Live event tail rendered into every `/` response when the agent is online.
+/// JS opens an `EventSource` on `/events/stream` and appends rows; no full-page
+/// reload, so the login flow and other forms aren't clobbered.
+const LIVE_PANEL: &str = r#"
+<h3>live</h3>
+<pre id="live" class="diff"><span class="meta">connecting…</span></pre>
+<script>
+(function() {
+  const log = document.getElementById('live');
+  function appendRow(text, cls) {
+    log.innerHTML = '';
+    const row = document.createElement('span');
+    if (cls) row.className = cls;
+    row.textContent = text + '\n';
+    log.appendChild(row);
+  }
+  function appendLine(text, cls) {
+    if (log.firstChild && log.firstChild.className === 'meta') log.innerHTML = '';
+    const row = document.createElement('span');
+    if (cls) row.className = cls;
+    row.textContent = text + '\n';
+    log.appendChild(row);
+    log.scrollTop = log.scrollHeight;
+  }
+  function fmt(ev) {
+    if (ev.kind === 'turn_start') return '◆ TURN ← ' + ev.from + ': ' + ev.body;
+    if (ev.kind === 'turn_end') return '◇ TURN END ' + (ev.ok ? 'ok' : 'fail') + (ev.note ? ' — ' + ev.note : '');
+    if (ev.kind === 'note') return '· ' + ev.text;
+    if (ev.kind === 'stream') {
+      // serde internal tagging flattens the inner json next to `kind`,
+      // so the original stream-json event sits under `ev` minus `kind`.
+      const v = Object.assign({}, ev); delete v.kind;
+      if (v.type === 'system' && v.subtype === 'init') return '[init] tools=' + (v.tools||[]).length;
+      if (v.type === 'assistant' && v.message && v.message.content) {
+        const parts = v.message.content.map(c => {
+          if (c.type === 'text') return c.text;
+          if (c.type === 'tool_use') return '→ ' + c.name + '(' + JSON.stringify(c.input) + ')';
+          return c.type;
+        });
+        return parts.join('\n');
+      }
+      if (v.type === 'user' && v.message && v.message.content) {
+        const parts = v.message.content.map(c => {
+          if (c.type === 'tool_result') {
+            const txt = Array.isArray(c.content) ? c.content.map(p => p.text || '').join(' ') : (c.content || '');
+            return '← ' + (txt.length > 200 ? txt.slice(0,200) + '…' : txt);
+          }
+          return c.type;
+        });
+        return parts.join('\n');
+      }
+      if (v.type === 'result') return '[done] ' + (v.subtype || '') + (v.is_error ? ' error' : '');
+      return JSON.stringify(v);
+    }
+    return JSON.stringify(ev);
+  }
+  function cls(ev) {
+    if (ev.kind === 'turn_start') return 'turnstart';
+    if (ev.kind === 'turn_end') return ev.ok ? 'turnok' : 'turnfail';
+    if (ev.kind === 'note') return 'meta';
+    return '';
+  }
+  const es = new EventSource('/events/stream');
+  es.onmessage = function(e) {
+    try {
+      const ev = JSON.parse(e.data);
+      appendLine(fmt(ev), cls(ev));
+    } catch (err) {
+      appendLine('[parse err] ' + e.data, 'meta');
+    }
+  };
+  es.onerror = function() { appendLine('[disconnected — retrying]', 'meta'); };
+})();
+</script>
+"#;
+
 fn render_needs_login_idle() -> String {
     "<p class=\"status-needs-login\">▓█▓▒░ NEEDS L0G1N ▓█▓▒░</p>\n<p>No Claude session in <code>~/.claude/</code>. The harness is up but the turn loop is paused until you log in.</p>\n<form method=\"POST\" action=\"/login/start\">\n  <button type=\"submit\" class=\"btn btn-login\">◆ ST4RT L0G1N</button>\n</form>\n<p class=\"meta\">Spawns <code>claude auth login</code> over plain stdio pipes. The OAuth URL will appear here when claude emits it; paste the resulting code back into the form below.</p>".into()
 }
@@ -104,6 +191,18 @@ fn render_login_in_progress(session: &Arc<LoginSession>) -> String {
     )
 }
 
+async fn events_stream(
+    State(state): State<AppState>,
+) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
+    let rx = state.bus.subscribe();
+    let stream = BroadcastStream::new(rx).filter_map(|res| {
+        let ev = res.ok()?;
+        let json = serde_json::to_string(&ev).ok()?;
+        Some(Ok(Event::default().data(json)))
+    });
+    Sse::new(stream).keep_alive(KeepAlive::default())
+}
+
 async fn post_login_start(State(state): State<AppState>) -> Response {
     drop_if_finished(&state.session);
     {
@@ -244,5 +343,10 @@ const STYLE: &str = r#"
     word-break: break-all;
     max-height: 30em;
   }
+  #live { max-height: 24em; overflow-y: auto; }
+  #live span { display: block; }
+  #live .turnstart { color: var(--amber); }
+  #live .turnok { color: var(--green); }
+  #live .turnfail { color: #ff6b6b; }
 </style>
 "#;