host.sock: push a live agent-status stream instead of poll-only

hive-c0re/src/server.rs

@@ -63,17 +63,43 @@ async fn handle(stream: UnixStream, coord: Arc<Coordinator>) -> Result<()> {
         if n == 0 {
             return Ok(());
         }
-        let resp = match serde_json::from_str::<HostRequest>(line.trim()) {
-            Ok(req) => dispatch(&req, coord.clone()).await,
-            Err(e) => HostResponse::error(format!("parse error: {e}")),
+        let req = match serde_json::from_str::<HostRequest>(line.trim()) {
+            Ok(req) => req,
+            Err(e) => {
+                write_response(
+                    &mut write,
+                    &HostResponse::error(format!("parse error: {e}")),
+                )
+                .await?;
+                continue;
+            }
         };
-        let mut payload = serde_json::to_string(&resp)?;
-        payload.push('\n');
-        write.write_all(payload.as_bytes()).await?;
-        write.flush().await?;
+        // The one verb that answers with more than one response — see its
+        // doc comment. Hands `write` over and stops reading further
+        // requests on this connection; a client wanting anything else
+        // opens a fresh one.
+        if matches!(req, HostRequest::SubscribeAgentStatus) {
+            return stream_agent_status(write, coord).await;
+        }
+        let resp = dispatch(&req, coord.clone()).await;
+        write_response(&mut write, &resp).await?;
     }
 }
 
+/// Serialize one `HostResponse` as a JSON line and flush it. Shared by the
+/// ordinary one-response-per-request path and `stream_agent_status`'s
+/// multi-response one, so the two can't quietly drift on framing.
+async fn write_response(
+    write: &mut tokio::net::unix::OwnedWriteHalf,
+    resp: &HostResponse,
+) -> Result<()> {
+    let mut payload = serde_json::to_string(resp)?;
+    payload.push('\n');
+    write.write_all(payload.as_bytes()).await?;
+    write.flush().await?;
+    Ok(())
+}
+
 #[allow(
     clippy::too_many_lines,
     reason = "flat one-arm-per-HostRequest-variant router; each arm just \
@@ -159,6 +185,13 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
             // see the request's doc comment for why the client asks.
             HostRequest::AgentExists { name } => HostResponse::agent_exists(agent_exists(name)?),
             HostRequest::AgentStatus => handle_agent_status(&coord).await,
+            // Intercepted in `handle()` before a request ever reaches
+            // `dispatch` — see `stream_agent_status`'s doc comment. This
+            // arm exists only so the match stays exhaustive; reaching it
+            // would mean a future caller invoked `dispatch` directly.
+            HostRequest::SubscribeAgentStatus => {
+                HostResponse::error("SubscribeAgentStatus is handled by the connection loop")
+            }
             // The hive domain + per-surface public URLs are injected into
             // c0re's service env by the hyperhive module; surface them so the
             // operator CLI can fill in this hive's own identity (the
@@ -368,30 +401,54 @@ async fn handle_agent_status(coord: &Arc<Coordinator>) -> HostResponse {
     let rows = crate::container_view::build_all(&coord.hive_env())
         .await
         .into_iter()
-        .map(|v| hive_sh4re::container::AgentStatusRow {
-            name: v.name,
-            running: v.running,
-            failed: v.failed,
-            needs_update: v.needs_update,
-            needs_login: v.needs_login,
-            deployed_sha: v.deployed_sha,
-            // Reminders are agent-local now; c0re has no cross-agent
-            // visibility into pending counts anymore. Stubbed
-            // to 0 rather than deleting the wire field outright — leaves
-            // `hivectl status`/the dashboard column intact syntactically,
-            // just always empty, until iris's frontend follow-up decides
-            // whether to drop the column entirely.
-            pending_reminders: 0,
-            parent: v.parent,
-            paused: v.paused,
-            active_model: v.active_model,
-            status_text: v.status_text,
-            status_set_at: v.status_set_at,
-        })
+        .map(hive_sh4re::container::AgentStatusRow::from)
         .collect();
     HostResponse::agent_statuses(rows)
 }
 
+/// `SubscribeAgentStatus` — ack once, then push one single-row
+/// [`HostResponse::agent_statuses`] per agent-status change for as long as
+/// the client stays connected. Consumes `write` (rather than borrowing it,
+/// like every other handler here) because this is the one verb that keeps
+/// writing after its own response, so `handle()` hands over ownership and
+/// stops reading further requests on this connection once it calls this.
+///
+/// Rides the same `Coordinator.dashboard_events` broadcast channel the
+/// dashboard's own SSE route reads (`dashboard/state_snapshot.rs`) —
+/// `SetStatus` already triggers a `rescan_containers_and_emit` on every
+/// status change (`socket_server/mod.rs::handle_set_status`), so that
+/// channel already carries every event this needs; no separate plumbing.
+async fn stream_agent_status(
+    mut write: tokio::net::unix::OwnedWriteHalf,
+    coord: Arc<Coordinator>,
+) -> Result<()> {
+    let mut events = coord.dashboard_subscribe();
+    write_response(&mut write, &HostResponse::success()).await?;
+    loop {
+        match events.recv().await {
+            Ok(crate::dashboard_events::DashboardEvent::ContainerStateChanged {
+                container,
+                ..
+            }) => {
+                let row = hive_sh4re::container::AgentStatusRow::from(container);
+                write_response(&mut write, &HostResponse::agent_statuses(vec![row])).await?;
+            }
+            // Every other event kind on this channel is dashboard-only
+            // (approvals, broker traffic, the queue, …) — nothing this
+            // subscriber asked for.
+            Ok(_) => {}
+            // Best-effort, same contract the dashboard's own live channel
+            // already has (see `HostRequest::SubscribeAgentStatus`'s doc
+            // comment) — a slow reader drops updates rather than stalling
+            // the broadcaster for everyone else on the channel.
+            Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
+                tracing::warn!(skipped, "agent-status subscribe: receiver lagged");
+            }
+            Err(tokio::sync::broadcast::error::RecvError::Closed) => return Ok(()),
+        }
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Matrix provisioning handlers
 //