From 1d902a099282d63ebe3a7f5dbc8cc037812ec835 Mon Sep 17 00:00:00 2001 From: iris Date: Sun, 13 Sep 2026 17:42:32 +0200 Subject: [PATCH 1/2] swarm-controller: add GET /api/agents/{hive}/{name}/term/stream hive-agent already publishes classified TermMsg rows to the core NATS subject $SWARM.term.{hive}.{agent} -- live only, no retention, by design. This endpoint subscribes that subject per request and relays each row over SSE, opaque to this daemon (no TermMsg dependency, same pass-through shape crate::status already uses for hive snapshots). No replay/history: the publish side never grew JetStream retention, and this route's own job (a live tail) never needed it. --- swarm-controller/src/main.rs | 2 + swarm-controller/src/term_stream.rs | 101 ++++++++++++++++++++++++++++ 2 files changed, 103 insertions(+) create mode 100644 swarm-controller/src/term_stream.rs diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 86c3778d..bd5c674c 100644 --- a/swarm-controller/src/main.rs +++ b/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)) diff --git a/swarm-controller/src/term_stream.rs b/swarm-controller/src/term_stream.rs new file mode 100644 index 00000000..0cd616b2 --- /dev/null +++ b/swarm-controller/src/term_stream.rs @@ -0,0 +1,101 @@ +//! `GET /api/agents/{hive}/{name}/term/stream` — a live SSE relay of one +//! agent's terminal rows, sourced straight from the swarm queue. +//! +//! **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 it would have on the agent's own +//! local SSE stream. This handler relays exactly that: no replay, no +//! last-N history. The publish side's own "last-N retention" half was +//! deliberately never built (a fresh-subscriber replay was reconsidered +//! and dropped once the publish path shipped without it) and this route +//! doesn't need it to do its own job, which is a live tail, not a +//! history endpoint. +//! +//! **The payload is passed through opaquely**, the same way +//! `crate::status::HiveStatus::snapshot` stores a hive's status +//! snapshot: this daemon has no reason to depend on `hive-agent`'s +//! `TermMsg` type just to re-serialise a row it never inspects. The +//! bytes on the wire are already what `hive-agent::swarm_term::publish` +//! sent, so decoding and re-encoding here would only be a chance to +//! disagree with it. + +use std::convert::Infallible; + +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..` 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/{hive}/{name}/term/stream", + params( + ("hive" = String, Path, description = "hive the agent runs on"), + ("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 hive or agent name is not shaped like an \ + identifier, or the hive is not in this swarm (problem+json)", body = String), + (status = 503, description = "no swarm queue is configured on this host, or it \ + could not be reached (problem+json)", body = String), + (status = 500, description = "the subscribe itself failed (problem+json)", body = String), + ), + tag = "agents" +)] +pub(crate) async fn stream_agent_term( + State(state): State, + Path((hive, agent)): Path<(String, String)>, +) -> Result>>, 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 hive = crate::swarm_hive(&state, &hive) + .map_err(|(code, detail)| crate::error_problem(code, &detail))?; + let agent = hive_types::Ident::parse(&agent) + .map_err(|reason| crate::error_problem(axum::http::StatusCode::BAD_REQUEST, reason))? + .into_string(); + + 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())) +} From c43a4577521dab19ed64ca14ccb38e8a705c7171 Mon Sep 17 00:00:00 2001 From: iris Date: Sun, 13 Sep 2026 17:47:14 +0200 Subject: [PATCH 2/2] 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/term_stream.rs | 85 ++++++++++++++++++++--------- 1 file changed, 58 insertions(+), 27 deletions(-) diff --git a/swarm-controller/src/term_stream.rs b/swarm-controller/src/term_stream.rs index 0cd616b2..628f364f 100644 --- a/swarm-controller/src/term_stream.rs +++ b/swarm-controller/src/term_stream.rs @@ -1,27 +1,30 @@ -//! `GET /api/agents/{hive}/{name}/term/stream` — a live SSE relay of one -//! agent's terminal rows, sourced straight from the swarm queue. +//! `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 it would have on the agent's own -//! local SSE stream. This handler relays exactly that: no replay, no -//! last-N history. The publish side's own "last-N retention" half was -//! deliberately never built (a fresh-subscriber replay was reconsidered -//! and dropped once the publish path shipped without it) and this route -//! doesn't need it to do its own job, which is a live tail, not a -//! history endpoint. +//! 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: this daemon has no reason to depend on `hive-agent`'s -//! `TermMsg` type just to re-serialise a row it never inspects. The -//! bytes on the wire are already what `hive-agent::swarm_term::publish` -//! sent, so decoding and re-encoding here would only be a chance to -//! disagree with it. +//! 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}; @@ -38,27 +41,28 @@ const SUBJECT_PREFIX: &str = "$SWARM.term"; #[utoipa::path( get, - path = "/api/agents/{hive}/{name}/term/stream", - params( - ("hive" = String, Path, description = "hive the agent runs on"), - ("name" = String, Path, description = "agent whose terminal to stream"), - ), + 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 hive or agent name is not shaped like an \ - identifier, or the hive is not in this swarm (problem+json)", body = String), - (status = 503, description = "no swarm queue is configured on this host, or it \ - could not be reached (problem+json)", body = String), - (status = 500, description = "the subscribe itself failed (problem+json)", body = String), + (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, - Path((hive, agent)): Path<(String, String)>, + Path(agent): Path, ) -> Result>>, problem_details::ProblemDetails> { let Some(status) = state.status.as_ref() else { return Err(crate::error_problem( @@ -66,12 +70,39 @@ pub(crate) async fn stream_agent_term( "no swarm queue is configured on this host", )); }; - let hive = crate::swarm_hive(&state, &hive) - .map_err(|(code, detail)| crate::error_problem(code, &detail))?; + 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