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.
This commit is contained in:
iris 2026-09-13 17:47:14 +02:00 committed by mara
commit c43a457752

View file

@ -1,27 +1,30 @@
//! `GET /api/agents/{hive}/{name}/term/stream` — a live SSE relay of one //! `GET /api/agents/{name}/term/stream` — a live SSE relay of one agent's
//! agent's terminal rows, sourced straight from the swarm queue. //! 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 //! **Live tail only, on purpose.** `hive-agent` publishes each
//! already-classified `TermMsg` row to the core subject //! already-classified `TermMsg` row to the core subject
//! `$SWARM.term.{hive}.{agent}` (see `hive-agent::swarm_term`'s module //! `$SWARM.term.{hive}.{agent}` (see `hive-agent::swarm_term`'s module
//! doc) — a core subject, not `JetStream`, so a subscriber who was not //! 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 //! listening missed the row, same as on the agent's own local SSE
//! local SSE stream. This handler relays exactly that: no replay, no //! stream. This handler relays exactly that: no replay, no last-N
//! last-N history. The publish side's own "last-N retention" half was //! history. The publish side's own retention half was deliberately
//! deliberately never built (a fresh-subscriber replay was reconsidered //! never built, and this route's own job — a live tail — never needed
//! and dropped once the publish path shipped without it) and this route //! it either.
//! 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 //! **The payload is passed through opaquely**, the same way
//! `crate::status::HiveStatus::snapshot` stores a hive's status //! `crate::status::HiveStatus::snapshot` stores a hive's status
//! snapshot: this daemon has no reason to depend on `hive-agent`'s //! snapshot: no reason to depend on `hive-agent`'s `TermMsg` type just
//! `TermMsg` type just to re-serialise a row it never inspects. The //! to re-serialise a row this daemon never inspects. The bytes on the
//! bytes on the wire are already what `hive-agent::swarm_term::publish` //! wire are already what `hive-agent::swarm_term::publish` sent.
//! sent, so decoding and re-encoding here would only be a chance to
//! disagree with it.
use std::convert::Infallible; use std::convert::Infallible;
use std::time::SystemTime;
use axum::extract::{Path, State}; use axum::extract::{Path, State};
use axum::response::sse::{Event, KeepAlive, Sse}; use axum::response::sse::{Event, KeepAlive, Sse};
@ -38,27 +41,28 @@ const SUBJECT_PREFIX: &str = "$SWARM.term";
#[utoipa::path( #[utoipa::path(
get, get,
path = "/api/agents/{hive}/{name}/term/stream", path = "/api/agents/{name}/term/stream",
params( params(("name" = String, Path, description = "agent whose terminal to stream")),
("hive" = String, Path, description = "hive the agent runs on"),
("name" = String, Path, description = "agent whose terminal to stream"),
),
responses( responses(
(status = 200, description = "server-sent event stream; each event's `data` is \ (status = 200, description = "server-sent event stream; each event's `data` is \
one already-classified TermMsg row, JSON as hive-agent published it \ one already-classified TermMsg row, JSON as hive-agent published it \
(opaque to this daemon) live only, no replay", (opaque to this daemon) live only, no replay",
body = String, content_type = "text/event-stream"), body = String, content_type = "text/event-stream"),
(status = 400, description = "the hive or agent name is not shaped like an \ (status = 400, description = "the agent name is not shaped like an identifier \
identifier, or the hive is not in this swarm (problem+json)", body = String), (problem+json)", body = String),
(status = 503, description = "no swarm queue is configured on this host, or it \ (status = 404, description = "the agent has never reported to the swarm, so no \
could not be reached (problem+json)", body = String), hive is on record to stream its terminal from (problem+json)", body = String),
(status = 500, description = "the subscribe itself failed (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" tag = "agents"
)] )]
pub(crate) async fn stream_agent_term( pub(crate) async fn stream_agent_term(
State(state): State<AppState>, State(state): State<AppState>,
Path((hive, agent)): Path<(String, String)>, Path(agent): Path<String>,
) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, problem_details::ProblemDetails> { ) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, problem_details::ProblemDetails> {
let Some(status) = state.status.as_ref() else { let Some(status) = state.status.as_ref() else {
return Err(crate::error_problem( return Err(crate::error_problem(
@ -66,12 +70,39 @@ pub(crate) async fn stream_agent_term(
"no swarm queue is configured on this host", "no swarm queue is configured on this host",
)); ));
}; };
let hive = crate::swarm_hive(&state, &hive) let Some(agent_status) = state.agent_status.as_ref() else {
.map_err(|(code, detail)| crate::error_problem(code, &detail))?; 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) let agent = hive_types::Ident::parse(&agent)
.map_err(|reason| crate::error_problem(axum::http::StatusCode::BAD_REQUEST, reason))? .map_err(|reason| crate::error_problem(axum::http::StatusCode::BAD_REQUEST, reason))?
.into_string(); .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(); let client = status.queue_client();
// Before subscribing: an unconnected client does not fail a // Before subscribing: an unconnected client does not fail a
// subscribe request outright, but there is no point opening a // subscribe request outright, but there is no point opening a