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.
This commit is contained in:
parent
77c3c656b2
commit
1d902a0992
2 changed files with 103 additions and 0 deletions
|
|
@ -50,6 +50,7 @@ mod otel_http_client;
|
||||||
mod read_policy;
|
mod read_policy;
|
||||||
mod status;
|
mod status;
|
||||||
mod store;
|
mod store;
|
||||||
|
mod term_stream;
|
||||||
mod vcs_metrics;
|
mod vcs_metrics;
|
||||||
mod wanted;
|
mod wanted;
|
||||||
mod webhook;
|
mod webhook;
|
||||||
|
|
@ -1674,6 +1675,7 @@ fn build_app(state: AppState) -> axum::Router {
|
||||||
.routes(routes!(set_agent_state))
|
.routes(routes!(set_agent_state))
|
||||||
.routes(routes!(matrix_account::put_matrix_account))
|
.routes(routes!(matrix_account::put_matrix_account))
|
||||||
.routes(routes!(get_hive_wanted))
|
.routes(routes!(get_hive_wanted))
|
||||||
|
.routes(routes!(term_stream::stream_agent_term))
|
||||||
.routes(routes!(issue_report::get_repos))
|
.routes(routes!(issue_report::get_repos))
|
||||||
.routes(routes!(issue_report::get_issue_report_all))
|
.routes(routes!(issue_report::get_issue_report_all))
|
||||||
.routes(routes!(issue_report::get_issue_report))
|
.routes(routes!(issue_report::get_issue_report))
|
||||||
|
|
|
||||||
101
swarm-controller/src/term_stream.rs
Normal file
101
swarm-controller/src/term_stream.rs
Normal file
|
|
@ -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.<hive>.<agent>` 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<AppState>,
|
||||||
|
Path((hive, agent)): Path<(String, String)>,
|
||||||
|
) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, 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()))
|
||||||
|
}
|
||||||
Loading…
Reference in a new issue