Compare commits
2 changed files with 0 additions and 134 deletions
|
|
@ -50,7 +50,6 @@ 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;
|
||||||
|
|
@ -1675,7 +1674,6 @@ 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))
|
||||||
|
|
|
||||||
|
|
@ -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.<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/{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<AppState>,
|
|
||||||
Path(agent): Path<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 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()))
|
|
||||||
}
|
|
||||||
Loading…
Reference in a new issue