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`]: the reasoning there for keeping the
//! hive out of the path applies unchanged here and is not repeated. Unlike
//! that route this one is a live tail with no replay, because no stream keeps
//! headers. The two subjects also differ in what they 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 and re-sends it once a minute, to a core
//! subject — so a client that attaches mid-idle sees nothing for up to a
//! minute. 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.
//!
//! **A client that attaches mid-idle gets the latest header at once.**
//! `crate::inference::Reported` keeps it, and the stream sends it first when
//! it is under [`REPLAY_MAX_AGE`] old: two re-send intervals, so an agent that
//! stopped does not keep a state it no longer has. After that it is the live
//! tail. A controller that has just started holds none and, as before, waits
//! for the agent's next header.
//!
//! **The payload is passed through opaquely**, same as the terminal relay:
//! this relay 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. [`crate::inference`]
//! reads its own fields out of the same subject separately.

use std::convert::Infallible;
use std::time::{Duration, SystemTime};

use axum::extract::{Path, State};
use axum::response::sse::{Event, KeepAlive, Sse};
use futures_util::{Stream, StreamExt as _, stream};

/// How old a header the stream still hands a new client first: two of the
/// agent's one-minute re-sends, so one missed re-send does not hide it.
const REPLAY_MAX_AGE: Duration = Duration::from_secs(150);

use crate::routes::AppState;

#[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 relay) — \
            `turn_state`, `turn_state_since` (ISO 8601 UTC), `agent_state`, `model`, `effort`, \
            `model_source`, `effort_source`, `resolved_model`, `context_window_tokens`, \
            `ctx_usage`, `cost_usage`, `needs_login`. Sent on transition and re-sent once a \
            minute, live. A new client is first sent the latest one the controller holds, \
            if it is under 150 seconds old",
            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_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::routes::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::routes::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::routes::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, "state stream: reading agent status failed");
            crate::routes::error_problem(axum::http::StatusCode::INTERNAL_SERVER_ERROR, &detail)
        })?
        .into_iter()
        .find_map(|row| row.hive)
        .ok_or_else(|| {
            crate::routes::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::routes::error_problem(
            axum::http::StatusCode::SERVICE_UNAVAILABLE,
            &swarm_queue_client::chain(&e),
        )
    })?;

    let subscriber = crate::term_stream::subscribe(
        &client,
        swarm_queue_client::agent_state_subject(&agent),
        "state",
    )
    .await?;
    // Read after subscribing, so a header published in between is in the tail
    // too; the client may see one twice, never miss one.
    let latest = state
        .reported_inference
        .as_ref()
        .and_then(|reported| reported.header_within(&agent, REPLAY_MAX_AGE));
    let first = stream::iter(latest.map(|payload| {
        Ok::<_, Infallible>(Event::default().data(String::from_utf8_lossy(&payload)))
    }));
    let live =
        subscriber.map(|msg| Ok(Event::default().data(String::from_utf8_lossy(&msg.payload))));
    let stream = first.chain(live);
    Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
}