hyperhive/swarm-controller/src/agent_state_stream.rs

//! `GET /api/agents/{name}/state/stream` — a live SSE relay of one agent's
//! turn-state header, sourced straight from the swarm queue.
//!
//! Sibling of [`crate::term_stream`] in every structural respect — the
//! reasoning there for resolving the hive at request time rather than
//! naming it in the path, and for a live tail with no replay, applies
//! unchanged here and is not repeated. What differs is only what the two
//! subjects carry: a terminal row is an event that happened, a header is
//! the agent's current state.
//!
//! **That difference is the one thing worth reading twice.** `hive-agent`
//! publishes a header on transition, to a core subject — so a client that
//! attaches mid-idle sees nothing until the next change, which for an agent
//! parked overnight can be hours. A renderer therefore has to open with
//! whatever `GET /api/agents/status` already gives it and let this stream
//! sharpen it, rather than treating an empty stream as an empty agent. The
//! alternative would be a KV bucket written on every turn-state flip, which
//! is the write rate this feature exists to avoid.
//!
//! **The payload is passed through opaquely**, same as the terminal relay:
//! this daemon never inspects a header, so it does not depend on
//! `hive-agent`'s type to re-serialise one. The bytes on the wire are what
//! `hive-agent::swarm_agent_state::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 turn-state headers — must match
/// `hive-agent::swarm_agent_state::SUBJECT_PREFIX` exactly, since the two
/// ends never see the constant together. `crate::term_stream::agent_subjects`
/// spells the full subjects; as with the terminal relay's own copy, 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.agent-state";

#[utoipa::path(
    get,
    path = "/api/agents/{name}/state/stream",
    params(("name" = String, Path, description = "agent whose turn state to stream")),
    responses(
        (status = 200, description = "server-sent event stream; each event's `data` is one \
            turn-state header, JSON as hive-agent published it (opaque to this daemon) — \
            `turn_state`, `turn_state_since` (ISO 8601 UTC), `agent_state`, `model`, \
            `resolved_model`, `context_window_tokens`, `ctx_usage`, `cost_usage`. Sent on \
            transition only, live, no replay: an idle agent emits nothing until it changes",
            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 state 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_state(
    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, "state 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 state 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 subscriber = crate::term_stream::subscribe_all(
        &client,
        &crate::term_stream::agent_subjects(SUBJECT_PREFIX, &hive, &agent),
        "state",
    )
    .await?;
    let stream =
        subscriber.map(|msg| Ok(Event::default().data(String::from_utf8_lossy(&msg.payload))));
    Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
}