hyperhive/swarm-controller/src/term_stream.rs

//! `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. `crate::agent_status`'s bucket is read only to answer
//! 404 for an agent that has never reported.
//!
//! **Live tail only, on purpose.** `hive-agent` publishes each
//! already-classified `TermMsg` row to the core subject
//! `$SWARM.term.{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. [`agent_subject`] spells 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 \
            (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 status 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.
    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"),
            )
        })?;

    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 subscriber = subscribe(&client, agent_subject(SUBJECT_PREFIX, &agent), "term").await?;
    let stream =
        subscriber.map(|msg| Ok(Event::default().data(String::from_utf8_lossy(&msg.payload))));
    Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
}

/// The subject one agent publishes on under `prefix`: `<prefix>.<agent>`.
pub(crate) fn agent_subject(prefix: &str, agent: &str) -> String {
    format!("{prefix}.{agent}")
}

/// Subscribe to `subject`. `what` names the relay in logs and errors.
pub(crate) async fn subscribe(
    client: &async_nats::Client,
    subject: String,
    what: &str,
) -> Result<async_nats::Subscriber, problem_details::ProblemDetails> {
    let subscriber = client.subscribe(subject.clone()).await.map_err(|e| {
        tracing::warn!(%subject, error = %e, "{what} stream: subscribe failed");
        crate::error_problem(
            axum::http::StatusCode::INTERNAL_SERVER_ERROR,
            &format!("subscribing to {subject} failed: {e}"),
        )
    })?;
    tracing::info!(%subject, "{what} stream: client attached");
    Ok(subscriber)
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The subject the agent's own queue credential is granted.
    #[test]
    fn a_relay_subscribes_to_the_agents_own_subject() {
        assert_eq!(agent_subject(SUBJECT_PREFIX, "atlas"), "$SWARM.term.atlas");
    }
}