From 1d902a099282d63ebe3a7f5dbc8cc037812ec835 Mon Sep 17 00:00:00 2001 From: iris Date: Sun, 13 Sep 2026 17:42:32 +0200 Subject: [PATCH] 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())) +}