swarm: publish each agent's turn-state header on its own subject

The swarm can already tell whether an agent is alive — the `agent-status`
KV bucket republishes once a minute — but not what it is doing right now.
A header bar wants the second thing, and a minute-old answer to "is this
agent thinking" is the wrong answer most of the time it is read.

`hive-agent` now publishes a turn-state header to
`$SWARM.agent-state.<hive>.<agent>`, a core subject beside the terminal
rows it already sends. It goes out **on transition, not on a timer**: the
publisher watches the event bus, rebuilds the header, and sends only when
the serialised result differs from the last one it sent — so a second
periodic writer, which is the problem this exists to fix, is not what
replaces the bucket.

The payload is the published contract a swarm-level renderer is written
against, so the test asserts on the serialised JSON keys rather than on
Rust field names. Two fields deliberately depart from the per-agent web
UI's `StateSnapshot`: `turn_state_since` is an ISO 8601 UTC string rather
than unix seconds, matching the sibling `$SWARM.term` subject's stamp, and
`agent_state` carries the swarm's own `AgentState` vocabulary rather than
a `paused` boolean, so a reader can compare actual against wanted without
translating. `turn_state` and `agent_state` stay two separate fields:
neither vocabulary contains the other's values.

Swarm-side, `GET /api/agents/{name}/state/stream` relays the subject as
SSE, resolving the agent's hive at request time exactly as the terminal
stream does and passing the bytes through without parsing them.

The broker grant is a second `--agent-publish-subject` rather than a
widening of the existing one, so the terminal family and the header family
stay independently revocable, and a `module-eval` arm pins the rendered
flag and its argument together — the doubled dollar included, since a
single one expands to nothing in `ExecStart` and yields a grant that
matches nothing.

Refs #3802

swarm-controller/src/agent_state_stream.rs

@@ -0,0 +1,133 @@
+//! `GET /api/agents/{name}/state/stream` — a live SSE relay of one agent's
+//! turn-state header, sourced straight from the swarm queue.
+//!
+//! Sibling of [`crate::term_stream`] in every structural respect — the
+//! reasoning there for resolving the hive at request time rather than
+//! naming it in the path, and for a live tail with no replay, applies
+//! unchanged here and is not repeated. What differs is only what the two
+//! subjects carry: a terminal row is an event that happened, a header is
+//! the agent's current state.
+//!
+//! **That difference is the one thing worth reading twice.** `hive-agent`
+//! publishes a header on transition, to a core subject — so a client that
+//! attaches mid-idle sees nothing until the next change, which for an agent
+//! parked overnight can be hours. A renderer therefore has to open with
+//! whatever `GET /api/agents/status` already gives it and let this stream
+//! sharpen it, rather than treating an empty stream as an empty agent. The
+//! alternative would be a KV bucket written on every turn-state flip, which
+//! is the write rate this feature exists to avoid.
+//!
+//! **The payload is passed through opaquely**, same as the terminal relay:
+//! this daemon never inspects a header, so it does not depend on
+//! `hive-agent`'s type to re-serialise one. The bytes on the wire are what
+//! `hive-agent::swarm_agent_state::publish` sent.
+
+use std::convert::Infallible;
+use std::time::SystemTime;
+
+use axum::extract::{Path, State};
+use axum::response::sse::{Event, KeepAlive, Sse};
+use futures_util::{Stream, StreamExt as _};
+
+use crate::AppState;
+
+/// Subject family carrying agent turn-state headers — must match
+/// `hive-agent::swarm_agent_state::SUBJECT_PREFIX` exactly, since the two
+/// ends never see the constant together. `$SWARM.agent-state.<hive>.<agent>`
+/// is the full subject; as with the terminal relay's own copy, there is no
+/// way to check the two sides agree short of this comment and the module
+/// docs on both ends staying honest about it.
+const SUBJECT_PREFIX: &str = "$SWARM.agent-state";
+
+#[utoipa::path(
+    get,
+    path = "/api/agents/{name}/state/stream",
+    params(("name" = String, Path, description = "agent whose turn state to stream")),
+    responses(
+        (status = 200, description = "server-sent event stream; each event's `data` is one \
+            turn-state header, JSON as hive-agent published it (opaque to this daemon) — \
+            `turn_state`, `turn_state_since` (ISO 8601 UTC), `agent_state`, `model`, \
+            `resolved_model`, `context_window_tokens`, `ctx_usage`, `cost_usage`. Sent on \
+            transition only, live, no replay: an idle agent emits nothing until it changes",
+            body = String, content_type = "text/event-stream"),
+        (status = 400, description = "the agent name is not shaped like an identifier \
+            (problem+json)", body = String),
+        (status = 404, description = "the agent has never reported to the swarm, so no \
+            hive is on record to stream its state from (problem+json)", body = String),
+        (status = 503, description = "no swarm queue is configured on this host, no \
+            agent-status reader is wired up, or the queue could not be reached \
+            (problem+json)", body = String),
+        (status = 500, description = "reading the agent's last-known hive failed, or the \
+            subscribe itself failed (problem+json)", body = String),
+    ),
+    tag = "agents"
+)]
+pub(crate) async fn stream_agent_state(
+    State(state): State<AppState>,
+    Path(agent): Path<String>,
+) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, problem_details::ProblemDetails> {
+    let Some(status) = state.status.as_ref() else {
+        return Err(crate::error_problem(
+            axum::http::StatusCode::SERVICE_UNAVAILABLE,
+            "no swarm queue is configured on this host",
+        ));
+    };
+    let Some(agent_status) = state.agent_status.as_ref() else {
+        return Err(crate::error_problem(
+            axum::http::StatusCode::SERVICE_UNAVAILABLE,
+            "no agent-status reader is configured on this host",
+        ));
+    };
+    let agent = hive_types::Ident::parse(&agent)
+        .map_err(|reason| crate::error_problem(axum::http::StatusCode::BAD_REQUEST, reason))?
+        .into_string();
+
+    // One-agent "roster": reuses `AgentStatusReader::view`'s already-tested
+    // row logic rather than a second, narrower lookup that could disagree
+    // with it about what "never reported" means.
+    let hive = agent_status
+        .view(std::slice::from_ref(&agent), SystemTime::now())
+        .await
+        .map_err(|e| {
+            let detail = format!("{e:#}");
+            tracing::warn!(%agent, error = %detail, "state stream: reading agent status failed");
+            crate::error_problem(axum::http::StatusCode::INTERNAL_SERVER_ERROR, &detail)
+        })?
+        .into_iter()
+        .find_map(|row| row.hive)
+        .ok_or_else(|| {
+            crate::error_problem(
+                axum::http::StatusCode::NOT_FOUND,
+                &format!(
+                    "{agent:?} has never reported to the swarm; no hive is on record to \
+                     stream its state from"
+                ),
+            )
+        })?;
+
+    let client = status.queue_client();
+    // Before subscribing: an unconnected client does not fail a subscribe
+    // request outright, but there is no point opening a subscription
+    // against a queue that is not there — the honest answer is 503, the
+    // same shape every other queue-backed route here already uses.
+    swarm_queue_client::ensure_connected(&client).map_err(|e| {
+        crate::error_problem(
+            axum::http::StatusCode::SERVICE_UNAVAILABLE,
+            &swarm_queue_client::chain(&e),
+        )
+    })?;
+
+    let subject = format!("{SUBJECT_PREFIX}.{hive}.{agent}");
+    let subscriber = client.subscribe(subject.clone()).await.map_err(|e| {
+        tracing::warn!(%subject, error = %e, "state stream: subscribe failed");
+        crate::error_problem(
+            axum::http::StatusCode::INTERNAL_SERVER_ERROR,
+            &format!("subscribing to {subject} failed: {e}"),
+        )
+    })?;
+
+    tracing::info!(%subject, "state stream: client attached");
+    let stream =
+        subscriber.map(|msg| Ok(Event::default().data(String::from_utf8_lossy(&msg.payload))));
+    Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
+}

swarm-controller/src/main.rs

@@ -40,6 +40,7 @@ use swarm_authelia_bridge_sock::BridgeResponse;
 use utoipa::{OpenApi, ToSchema};
 use utoipa_axum::{router::OpenApiRouter, routes};
 
+mod agent_state_stream;
 mod agent_status;
 mod auth;
 mod config_pr;
@@ -1676,6 +1677,7 @@ fn build_app(state: AppState) -> axum::Router {
         .routes(routes!(matrix_account::put_matrix_account))
         .routes(routes!(get_hive_wanted))
         .routes(routes!(term_stream::stream_agent_term))
+        .routes(routes!(agent_state_stream::stream_agent_state))
         .routes(routes!(issue_report::get_repos))
         .routes(routes!(issue_report::get_issue_report_all))
         .routes(routes!(issue_report::get_issue_report))