diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index bd5c674c..86c3778d 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -50,7 +50,6 @@ mod otel_http_client; mod read_policy; mod status; mod store; -mod term_stream; mod vcs_metrics; mod wanted; mod webhook; @@ -1675,7 +1674,6 @@ 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 deleted file mode 100644 index 628f364f..00000000 --- a/swarm-controller/src/term_stream.rs +++ /dev/null @@ -1,132 +0,0 @@ -//! `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())) -}