term_stream: drop hive from the URL, resolve it from agent_status

mara's review point on #4351: an agent isn't pinned to a hive forever
(it can move), so a URL naming one would go stale the moment it did.
Resolve the hive at request time from the agent-status bucket instead
-- the same source AgentStatusRow.hive already comes from -- rather
than trusting a caller-supplied value. Route is now
GET /api/agents/{name}/term/stream; a never-reported agent now answers
404 (no hive on record) instead of silently guessing.

swarm-controller/src/main.rs

@@ -50,6 +50,7 @@ mod otel_http_client;
 mod read_policy;
 mod status;
 mod store;
+mod term_stream;
 mod vcs_metrics;
 mod wanted;
 mod webhook;
@@ -1674,6 +1675,7 @@ fn build_app(state: AppState) -> axum::Router {
         .routes(routes!(set_agent_state))
         .routes(routes!(matrix_account::put_matrix_account))
         .routes(routes!(get_hive_wanted))
+        .routes(routes!(term_stream::stream_agent_term))
         .routes(routes!(issue_report::get_repos))
         .routes(routes!(issue_report::get_issue_report_all))
         .routes(routes!(issue_report::get_issue_report))

swarm-controller/src/term_stream.rs

@@ -0,0 +1,132 @@
+//! `GET /api/agents/{name}/term/stream` — a live SSE relay of one agent's
+//! terminal rows, sourced straight from the swarm queue.
+//!
+//! **No `hive` in the path, on purpose.** An agent is not pinned to a
+//! hive forever — it can move — so a URL naming one would go stale the
+//! moment it did. The hive is resolved at request time from
+//! `crate::agent_status`'s bucket instead, the same "an agent's hive
+//! comes from its own last report" rule that module already documents.
+//!
+//! **Live tail only, on purpose.** `hive-agent` publishes each
+//! already-classified `TermMsg` row to the core subject
+//! `$SWARM.term.{hive}.{agent}` (see `hive-agent::swarm_term`'s module
+//! doc) — a core subject, not `JetStream`, so a subscriber who was not
+//! listening missed the row, same as on the agent's own local SSE
+//! stream. This handler relays exactly that: no replay, no last-N
+//! history. The publish side's own retention half was deliberately
+//! never built, and this route's own job — a live tail — never needed
+//! it either.
+//!
+//! **The payload is passed through opaquely**, the same way
+//! `crate::status::HiveStatus::snapshot` stores a hive's status
+//! snapshot: no reason to depend on `hive-agent`'s `TermMsg` type just
+//! to re-serialise a row this daemon never inspects. The bytes on the
+//! wire are already what `hive-agent::swarm_term::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 terminal rows — must match
+/// `hive-agent::swarm_term::SUBJECT_PREFIX` exactly, since the two ends
+/// never see the constant together. `$SWARM.term.<hive>.<agent>` is the
+/// full subject; 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.term";
+
+#[utoipa::path(
+    get,
+    path = "/api/agents/{name}/term/stream",
+    params(("name" = String, Path, description = "agent whose terminal to stream")),
+    responses(
+        (status = 200, description = "server-sent event stream; each event's `data` is \
+            one already-classified TermMsg row, JSON as hive-agent published it \
+            (opaque to this daemon) — live only, no replay",
+            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 terminal 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_term(
+    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, "term 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 terminal 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, "term stream: subscribe failed");
+        crate::error_problem(
+            axum::http::StatusCode::INTERNAL_SERVER_ERROR,
+            &format!("subscribing to {subject} failed: {e}"),
+        )
+    })?;
+
+    tracing::info!(%subject, "term 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()))
+}