From c43a4577521dab19ed64ca14ccb38e8a705c7171 Mon Sep 17 00:00:00 2001 From: iris Date: Sun, 13 Sep 2026 17:47:14 +0200 Subject: [PATCH] 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