hyperhive/hive-agent/src/swarm_term.rs

//! Publishing this agent's terminal rows onto the swarm queue.
//!
//! The per-agent web UI already classifies the event bus into [`TermMsg`]
//! rows; this offers the same rows upward so a swarm-level terminal can
//! render an agent without reaching into its hive. It publishes the value the
//! web UI is handed rather than a second model, so a new tool or a
//! reclassified event changes both surfaces at once.
//!
//! **Subscribes to the bus, not to a browser.** The SSE handler builds its
//! rows per connected client, so hanging this off it would mean an agent
//! whose terminal nobody has open publishes nothing. This is its own
//! subscriber with its own long-lived [`ClassifyCtx`] — a publisher that
//! restarted its correlation state per reader would lose the `tool_use` →
//! name mapping a `tool_result` needs.
//!
//! **A core subject, not `JetStream`.** A terminal row has no current value a
//! late reader can ask for, and nothing here is worth the durability of the
//! notices stream: a subscriber that was not listening missed the row, the
//! same way it would have missed it on the agent's own SSE stream. That also
//! keeps the agent's grant to a plain publish.
//!
//! **Best-effort per row.** The turn loop and the web UI never wait on the
//! queue, so the bus receiver is the only thing this task blocks on; a failed
//! publish is a log line and the next row is still attempted.

use tokio::sync::broadcast;

use crate::events::BusEvent;
use crate::term_msg::{ClassifyCtx, TermMsg, classify};
use hive_sh4re::term_msg::fit;

/// How much room to leave under the announced limit for everything the
/// publish adds around the payload — subject, headers, protocol framing.
///
/// The server measures the payload alone, so this is not required for
/// correctness; it is here because the cost of being one byte over is not one
/// lost byte but a refused message and a dropped connection, and a kilobyte
/// out of eight megabytes buys that margin for nothing.
const HEADROOM: usize = 1024;

/// Start the publish task, if this agent has a label to name its subject with.
pub fn spawn(rx: broadcast::Receiver<BusEvent>) {
    let agent = crate::identity::label();
    if agent.is_empty() {
        tracing::warn!("this agent has no label; not publishing the terminal upward");
        return;
    }
    tokio::spawn(run(rx, agent));
}

async fn run(mut rx: broadcast::Receiver<BusEvent>, agent: String) {
    let client = crate::swarm_queue::client().await;
    let subject = swarm_queue_client::term_subject(&agent);
    tracing::info!(subject, "publishing the agent terminal to the swarm queue");

    // One context for the task's whole life — see the module doc.
    let mut ctx = ClassifyCtx::default();
    loop {
        let event = match rx.recv().await {
            Ok(event) => event,
            // The bus drops events for a subscriber that falls behind. The
            // terminal is a tail rather than a log, so the answer is to say
            // how many were missed and keep reading, exactly as the web UI's
            // own subscriber does.
            Err(broadcast::error::RecvError::Lagged(missed)) => {
                tracing::warn!(missed, "swarm terminal: lagged, rows skipped");
                continue;
            }
            // Every sender is gone, so the harness is shutting down.
            Err(broadcast::error::RecvError::Closed) => return,
        };
        for msg in classify(&event.event, event.ts, &mut ctx) {
            publish(&client, &subject, msg).await;
        }
    }
}

/// Offer one row. Every failure is terminal for that row and for nothing else.
async fn publish(client: &async_nats::Client, subject: &str, msg: TermMsg) {
    // An unconnected client does not fail a publish, it buffers it — and the
    // limit read below is the library's pre-connect default until the server
    // has announced its own, which is smaller than any deployment sets and
    // would degrade rows that would have fit.
    if let Err(e) = swarm_queue_client::ensure_connected(client) {
        tracing::warn!(error = %swarm_queue_client::chain(&e), "swarm terminal: publish skipped");
        return;
    }
    let limit = swarm_queue_client::max_payload(client).saturating_sub(HEADROOM);
    let summary = msg.summary.clone();
    let Some(msg) = fit(msg, limit) else {
        tracing::warn!(
            summary,
            limit,
            "swarm terminal: row does not fit even without its body, dropped"
        );
        return;
    };
    let payload = match serde_json::to_vec(&msg) {
        Ok(payload) => payload,
        Err(e) => {
            tracing::warn!(error = %e, "swarm terminal: serialising failed, row dropped");
            return;
        }
    };
    if let Err(e) = client.publish(subject.to_owned(), payload.into()).await {
        tracing::warn!(error = %e, "swarm terminal: publish failed, row dropped");
    }
}

#[cfg(test)]
mod tests {
    use crate::events::LiveEvent;
    use crate::term_msg::{ClassifyCtx, classify};
    use hive_sh4re::term_msg::fit;

    /// The time `classify` renders for the event the test below classifies.
    const ROW_TS: &str = "2026-09-13T12:35:03Z";

    /// What a subscriber actually receives: the payload this module hands
    /// `publish` is the bare row, so the time has to be *in* it. A `ts` that
    /// existed in Rust but never serialized would leave the queue exactly as
    /// timeless as it was before.
    #[test]
    fn the_published_payload_carries_the_events_own_time() {
        let ev = LiveEvent::TurnEnd {
            ok: true,
            note: None,
        };
        let mut ctx = ClassifyCtx::default();
        // 2026-09-13T12:35:03Z, a time that is not the time of this run.
        let msgs = classify(&ev, 1_789_302_903, &mut ctx);
        let payload = serde_json::to_vec(&fit(msgs[0].clone(), 4096).expect("a small row fits"))
            .expect("serialises");
        let value: serde_json::Value = serde_json::from_slice(&payload).expect("valid json");
        assert_eq!(value["ts"], serde_json::json!(ROW_TS));
    }
}