From 3871749da93a56c3ba383f754b95e8228b02dd2e Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 7 Sep 2026 18:24:22 +0200 Subject: [PATCH] host.sock: push a live agent-status stream instead of poll-only --- hive-c0re/src/container_view.rs | 81 +++++++++++++++++++++++ hive-c0re/src/server.rs | 111 ++++++++++++++++++++++++-------- hive-host-sock/src/lib.rs | 16 +++++ 3 files changed, 181 insertions(+), 27 deletions(-) diff --git a/hive-c0re/src/container_view.rs b/hive-c0re/src/container_view.rs index f0639e9e..a0d0afb3 100644 --- a/hive-c0re/src/container_view.rs +++ b/hive-c0re/src/container_view.rs @@ -96,6 +96,38 @@ pub struct ContainerView { pub memory_max: String, } +impl From for hive_sh4re::container::AgentStatusRow { + /// The one place a `ContainerView` becomes the wire row `host.sock` + /// serves it as — `handle_agent_status`'s poll path and the + /// `SubscribeAgentStatus` push path both go through this, so the two + /// can never quietly diverge on which fields make the cut. + /// + /// `pending_reminders` has no source here: reminders are agent-local + /// now, and c0re has no cross-agent visibility into pending counts + /// anymore. Stubbed to `0` rather than dropping 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. `port`, `cpu_quota`, + /// `memory_max` have no equivalent on the row at all and are simply + /// not carried over. + fn from(v: ContainerView) -> Self { + Self { + name: v.name, + running: v.running, + failed: v.failed, + needs_update: v.needs_update, + needs_login: v.needs_login, + deployed_sha: v.deployed_sha, + 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, + } + } +} + /// Build the full container list. Wraps `lifecycle::list()` and /// resolves every per-agent attribute the dashboard surfaces. /// @@ -468,4 +500,53 @@ mod tests { let ok = lock(r#""a":"a""#, r#""a":{"locked":{"rev":"abc"}}"#); assert_eq!(parse_locked_revs(&ok).len(), 1); } + + /// The `From` impl both `handle_agent_status`'s poll path and + /// `stream_agent_status`'s push path go through — pins the field + /// mapping so the two can't quietly diverge, and that `port` / + /// `cpu_quota` / `memory_max` (no equivalent on the row) are dropped + /// on purpose rather than by omission. + #[test] + fn agent_status_row_carries_every_field_the_row_has_and_drops_the_rest() { + use super::ContainerView; + use chrono::{TimeZone, Utc}; + use hive_sh4re::container::AgentStatusRow; + + let view = ContainerView { + name: "alice".to_owned(), + container: "h-alice".to_owned(), + port: 7000, + running: true, + failed: false, + needs_update: true, + needs_login: false, + deployed_sha: Some("abc123def456".to_owned()), + parent: Some("bob".to_owned()), + active_model: Some("claude-opus".to_owned()), + status_text: Some("shipping".to_owned()), + status_set_at: Some(Utc.timestamp_opt(1_700_000_000, 0).unwrap()), + paused: true, + cpu_quota: "400%".to_owned(), + memory_max: "8G".to_owned(), + }; + + let row = AgentStatusRow::from(view); + assert_eq!(row.name, "alice"); + assert!(row.running); + assert!(!row.failed); + assert!(row.needs_update); + assert!(!row.needs_login); + assert_eq!(row.deployed_sha.as_deref(), Some("abc123def456")); + assert_eq!(row.parent.as_deref(), Some("bob")); + assert!(row.paused); + assert_eq!(row.active_model.as_deref(), Some("claude-opus")); + assert_eq!(row.status_text.as_deref(), Some("shipping")); + assert_eq!( + row.status_set_at, + Some(Utc.timestamp_opt(1_700_000_000, 0).unwrap()) + ); + // No source for reminders on this side (see the `From` impl's doc + // comment) — always the stub, never left uninitialised. + assert_eq!(row.pending_reminders, 0); + } } diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 9eaba30e..67af8b76 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -63,17 +63,43 @@ async fn handle(stream: UnixStream, coord: Arc) -> Result<()> { if n == 0 { return Ok(()); } - let resp = match serde_json::from_str::(line.trim()) { - Ok(req) => dispatch(&req, coord.clone()).await, - Err(e) => HostResponse::error(format!("parse error: {e}")), + let req = match serde_json::from_str::(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) -> 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) -> 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, +) -> 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 // diff --git a/hive-host-sock/src/lib.rs b/hive-host-sock/src/lib.rs index 29dc777c..496afed2 100644 --- a/hive-host-sock/src/lib.rs +++ b/hive-host-sock/src/lib.rs @@ -206,6 +206,22 @@ pub enum HostRequest { /// pending reminders) — the `hivectl list-agents` roster view. /// Reuses the dashboard's per-agent `ContainerView` aggregation. AgentStatus, + /// Turn this connection into a live push feed of agent-status changes, + /// so a polling loop against [`HostRequest::AgentStatus`] is no longer + /// the only way to notice one. The one verb on this socket that answers + /// with more than one response: the server acks with a bare + /// [`HostResponse::success()`] first, then keeps the connection open + /// and writes one further [`HostResponse::agent_statuses`] (always a + /// single-row `Vec`) per changed agent, for as long as the client stays + /// connected. No further requests are read on this connection after + /// this one — open a fresh connection for anything else. + /// + /// Best-effort, same contract the dashboard's own live channel already + /// has: a slow reader can miss updates rather than back-pressuring the + /// daemon, so a client that cares about the *current* state after a gap + /// should still fall back to a fresh [`HostRequest::AgentStatus`] poll, + /// not assume the stream alone is authoritative. + SubscribeAgentStatus, /// Report this hive's canonical DNS domain /// (`services.hyperhive.domain`) plus the browser-facing home / /// forge / matrix URLs, daemon-sourced so custom forge/matrix