//! 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 in every direction.** The turn loop and the web UI must not
//! notice whether the queue exists, 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};

/// Subject family carrying agent terminal rows, the swarm-wide agreement this
/// publisher holds up its end of. One leaf subject per agent, so a subscriber
/// can follow one agent without filtering the swarm's whole terminal traffic.
const SUBJECT_PREFIX: &str = "$SWARM.term";

/// What the queue's client id looks like either side of the hive's own name.
///
/// These mirror the auth-callout responder's `--hive-client-prefix` and
/// `--agent-client-suffix`, which is where the authoritative pair lives: the
/// responder parses the hive back out of the presented client id and grants
/// exactly `$SWARM.term.<that hive>.>`. The agent is not told those flags, so
/// it restates their defaults — a deployment that retunes either one has to
/// change them here too, and the symptom of not doing so is every publish
/// refused rather than a wrong subject being accepted.
const CLIENT_ID_PREFIX: &str = "hive-";
const CLIENT_ID_SUFFIX: &str = "-agent";

/// What a row that lost its body to the payload limit carries instead.
const DROPPED_BODY: &str = "[body dropped: over the queue's payload limit]";

/// 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;

/// The hive a queue client id names.
///
/// The hive in the subject has to be the string the **responder** parses out
/// of this same client id, because that is what its grant is built from. The
/// harness knows a hive display name too, from a different source and with no
/// rule tying the two together — deriving the subject from that one instead
/// produces a publish the broker refuses, which surfaces as a terminal that is
/// simply empty at the swarm.
///
/// `None` for an id that does not have the expected shape: a publisher that
/// guessed at a subject would be asking for a grant it cannot have, so the
/// caller disables itself instead.
fn hive_from_client_id(client_id: &str) -> Option<&str> {
    let hive = client_id
        .strip_prefix(CLIENT_ID_PREFIX)?
        .strip_suffix(CLIENT_ID_SUFFIX)?;
    (!hive.is_empty()).then_some(hive)
}

/// Bring `msg` under `limit` serialized bytes, or report that it cannot be.
///
/// The body is the only field that carries arbitrary length — a diff, a whole
/// tool result — so it is the only one worth spending: replacing it keeps the
/// row's identity, level, summary and time, which is what makes the row
/// readable at all, and a reader sees that something was there rather than
/// seeing nothing.
///
/// `None` means even the degraded row does not fit, so the caller reports it
/// instead of publishing. That matters more than it looks: an oversize publish
/// is not truncated by the server, it is refused and the connection is closed,
/// which costs the row *and* every row racing behind it through the reconnect.
fn fit(msg: TermMsg, limit: usize) -> Option<TermMsg> {
    if serialized_len(&msg)? <= limit {
        return Some(msg);
    }
    // Rebuilt rather than mutated in place: `body_format` describes the body,
    // and a marker left tagged `Diff` renders as a broken diff downstream.
    let degraded = TermMsg {
        body: Some(DROPPED_BODY.to_owned()),
        body_format: None,
        ..msg
    };
    (serialized_len(&degraded)? <= limit).then_some(degraded)
}

/// Serialized size of a row, or `None` if it does not serialize at all.
///
/// Measured by serializing rather than estimated from field lengths: the
/// payload is what the server measures, and JSON escaping makes the two differ
/// by an unbounded factor on exactly the rows that are already near the limit.
fn serialized_len(msg: &TermMsg) -> Option<usize> {
    serde_json::to_vec(msg).ok().map(|v| v.len())
}

/// Start the publish task, if this agent has both queue coordinates and an
/// identity the subject can be derived from.
///
/// Returns without spawning in every other case — no queue, an unparseable
/// client id, no label — each of which is a legal state for an agent rather
/// than an error, and each logged once here rather than per row.
pub fn spawn(rx: broadcast::Receiver<BusEvent>) {
    let Some(cfg) = crate::swarm_queue::config() else {
        // `swarm_queue::init` already said why at boot; repeating it here
        // would be the same fact logged twice.
        return;
    };
    let Some(hive) = hive_from_client_id(&cfg.client_id) else {
        tracing::warn!(
            client_id = %cfg.client_id,
            expected = format!("{CLIENT_ID_PREFIX}<hive>{CLIENT_ID_SUFFIX}"),
            "queue client id does not name a hive; not publishing the terminal upward"
        );
        return;
    };
    let agent = crate::identity::label();
    if agent.is_empty() {
        tracing::warn!("this agent has no label; not publishing the terminal upward");
        return;
    }
    let subject = format!("{SUBJECT_PREFIX}.{hive}.{agent}");
    tokio::spawn(run(rx, subject));
}

async fn run(mut rx: broadcast::Receiver<BusEvent>, subject: String) {
    let Some(client) = crate::swarm_queue::client().await else {
        return;
    };
    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 super::{DROPPED_BODY, fit, hive_from_client_id};
    use crate::events::LiveEvent;
    use crate::term_msg::{BodyFormat, ClassifyCtx, Level, TermMsg, classify};

    /// A row's time as `classify` renders it, for the tests below that need
    /// one without classifying an event to get it.
    const ROW_TS: &str = "2026-09-13T12:35:03Z";

    #[test]
    fn a_client_id_names_the_hive_between_the_prefix_and_the_suffix() {
        assert_eq!(hive_from_client_id("hive-alpha-agent"), Some("alpha"));
        // A hive whose own name contains the suffix still resolves: only the
        // trailing one is stripped, so the responder and this agree.
        assert_eq!(
            hive_from_client_id("hive-alpha-agent-agent"),
            Some("alpha-agent")
        );
    }

    /// Each way an id can fail to name a hive. A guessed subject would be
    /// refused by the broker, and a refusal reaches an operator as an empty
    /// terminal rather than as an error, so none of these may fall back.
    #[test]
    fn an_id_of_another_shape_names_no_hive() {
        for id in [
            // The hive's own id, not an agent's.
            "hive-alpha",
            // Missing the prefix the responder keys on.
            "alpha-agent",
            // Prefix and suffix but nothing between them: the subject would
            // be `$SWARM.term..<agent>`, whose empty token matches no grant.
            "hive--agent",
            // Prefix and suffix overlapping with no hive at all.
            "hive-agent",
            "",
        ] {
            assert_eq!(hive_from_client_id(id), None, "{id} must name no hive");
        }
    }

    #[test]
    fn a_row_that_already_fits_is_published_unchanged() {
        let msg = TermMsg::new(Level::Info, "turn ok")
            .icon("✅")
            .body("a short body", Some(BodyFormat::Markdown));
        let fitted = fit(msg, 4096).expect("a small row fits");
        assert_eq!(fitted.body.as_deref(), Some("a short body"));
        assert_eq!(fitted.body_format, Some(BodyFormat::Markdown));
    }

    /// The case the degrade exists for: a body larger than the limit costs the
    /// body and nothing else, and what comes back is actually under the limit
    /// rather than merely smaller.
    #[test]
    fn an_oversize_body_is_replaced_and_the_result_fits() {
        let limit = 512;
        let msg = TermMsg::new(Level::Info, "Edit(src/main.rs)")
            .icon("🔧")
            .body("x".repeat(limit * 4), Some(BodyFormat::Diff))
            .coalesce("tool-1")
            .at(ROW_TS);
        let fitted = fit(msg, limit).expect("dropping the body brings this under the limit");
        assert_eq!(fitted.body.as_deref(), Some(DROPPED_BODY));
        // The tag describes a body that is no longer there; left set, a reader
        // renders the marker as a diff.
        assert_eq!(fitted.body_format, None);
        // The fields that make the row readable survive.
        assert_eq!(fitted.summary, "Edit(src/main.rs)");
        assert_eq!(fitted.icon.as_deref(), Some("🔧"));
        assert_eq!(fitted.coalesce_key.as_deref(), Some("tool-1"));
        // Including the time: a degraded row that lost it would be a row a
        // subscriber cannot place, and nothing downstream could tell.
        assert_eq!(fitted.ts, ROW_TS);
        assert!(
            serde_json::to_vec(&fitted).expect("serialises").len() <= limit,
            "the degraded row must be under the limit, not merely smaller"
        );
    }

    /// A row whose summary alone exceeds the limit cannot be degraded into
    /// one, and publishing it anyway would cost the connection rather than the
    /// row. Reported by the caller, never sent.
    #[test]
    fn a_row_too_large_even_without_its_body_is_refused() {
        let limit = 256;
        let msg = TermMsg::new(Level::Warn, "s".repeat(limit * 4)).body("x".repeat(limit), None);
        assert!(fit(msg, limit).is_none());
    }

    /// Control for the test above: the same oversize summary with a body that
    /// would fit still refuses, so the refusal is about total size and not
    /// about the body having been present.
    #[test]
    fn the_refusal_is_about_size_rather_than_the_body_being_present() {
        let limit = 256;
        let msg = TermMsg::new(Level::Warn, "s".repeat(limit * 4));
        assert!(fit(msg, limit).is_none());
    }

    /// JSON escaping is why the size is measured by serializing: a body of
    /// quotes serializes to twice its own length, so a row that fits by
    /// character count can still be refused on the wire.
    #[test]
    fn the_limit_is_measured_on_the_serialized_bytes() {
        let body = "\"".repeat(200);
        let msg = TermMsg::new(Level::Info, "quotes").body(body.clone(), None);
        let limit = body.len() + 64;
        // Well under the limit as characters, over it once escaped.
        assert!(
            serde_json::to_vec(&msg).expect("serialises").len() > limit,
            "this fixture must be oversize only after escaping"
        );
        let fitted = fit(msg, limit).expect("dropping the body fits");
        assert_eq!(fitted.body.as_deref(), Some(DROPPED_BODY));
    }

    /// 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));
    }
}