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..628f364f --- /dev/null +++ b/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..` 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, + Path(agent): Path, +) -> 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 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())) +}