//! Publishing this agent's turn-state header onto the swarm queue.
//!
//! The per-agent web UI's `/api/state` already carries what a header bar
//! wants — what the turn loop is doing, since when, on which model, how much
//! context and cost the last turn spent. None of it reaches the swarm: the
//! `agent-status` KV bucket republishes once a minute, which is fine for
//! "is this agent alive" and useless for "is it thinking right now". This
//! offers the same values upward on their own subject so a swarm-level
//! header can render an agent without reaching into its hive.
//!
//! **Published on transition.** The whole reason this exists is that the
//! once-a-minute bucket is too stale for "is it thinking right now". This
//! subscribes to the event bus and republishes whenever the snapshot it
//! builds differs from the one it last sent — see [`run`] for why the
//! comparison is on the serialized bytes rather than on an event allow-list.
//!
//! **A core subject, not `JetStream`**, for the reason
//! [`crate::swarm_term`] gives at more length: nothing here is worth the
//! durability of the notices stream, and it keeps the agent's grant to a
//! plain publish. The queue keeps no header for a late subscriber, so the
//! current header is also re-sent every [`REFRESH_INTERVAL`]: a reader that
//! attaches while the agent is idle (swarm-controller after a restart) has
//! it within that interval. Transitions still go out as they happen.
//!
//! **Best-effort per transition.** 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 transition is still attempted.

use std::sync::{Arc, Mutex, PoisonError};
use std::time::Duration;

use serde::Serialize;
use tokio::sync::broadcast;
use tokio::time::{Interval, MissedTickBehavior};

use swarm_wire::agent_inference::Effective;
use swarm_wire::agent_login::Login;
use swarm_wire::wanted::AgentState;

use crate::events::{Bus, BusEvent, LiveEvent, TurnState};
use crate::login::LoginState;
use crate::term_msg::iso8601_utc;

/// How much room to leave under the announced limit for everything the
/// publish adds around the payload — subject, headers, protocol framing.
/// Same margin and same reasoning as `swarm_term::HEADROOM`.
const HEADROOM: usize = 1024;

/// How often the current header is re-sent when nothing changed it.
const REFRESH_INTERVAL: Duration = Duration::from_mins(1);

/// One turn-state header, as a swarm-level reader receives it.
///
/// The field names here are the published contract; two of them
/// deliberately do **not** match the per-agent web UI's `StateSnapshot`,
/// which is the other place these same values are served from:
///
/// - `turn_state_since` is an ISO 8601 / RFC 3339 UTC string, where
///   `StateSnapshot` sends unix seconds. The sibling subject's
///   [`crate::term_msg::TermMsg::ts`] is already spelled that way, and two
///   messages on adjacent subjects disagreeing about how to write a time is
///   a cost paid by every reader of both.
/// - `agent_state` replaces `StateSnapshot`'s `paused: bool` with the
///   swarm's own [`AgentState`] — the same enum the *wanted* state is
///   declared in, so a reader can compare actual against wanted directly
///   instead of translating one into the other's vocabulary first.
///
/// `turn_state` and `agent_state` are two axes, not one: [`AgentState`] has
/// no notion of `thinking` and [`TurnState`] has no notion of `offline`, so
/// a single field would have to drop one of them.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
struct AgentStateMsg {
    /// What the turn loop is doing right now.
    turn_state: TurnState,
    /// When it entered that state, ISO 8601 / RFC 3339 UTC — see the struct
    /// doc. A reader ticks the elapsed time off this rather than being sent
    /// an age that is wrong the moment it is serialized.
    turn_state_since: String,
    /// What this agent can honestly say it *is* — see [`observed_state`],
    /// which documents why only two of the four variants can ever appear
    /// here.
    agent_state: AgentState,
    /// `model`, `effort`, `model_source` and `effort_source`: what the next
    /// turn runs on and where each comes from. swarm-controller decodes
    /// these four out of the header with the same type.
    #[serde(flatten)]
    inference: Effective,
    /// The concrete model id the most recent completed turn actually ran
    /// on, resolved from that turn's assistant events. `None` before any
    /// turn has completed.
    resolved_model: Option<String>,
    /// Effective context-window token budget for the current model.
    context_window_tokens: u64,
    /// Last-inference token usage from the most recent completed turn — the
    /// current context-window occupancy. `None` until the first turn ends.
    ctx_usage: Option<hive_claude::TokenUsage>,
    /// Cumulative token usage across the most recent turn's inferences, the
    /// cost signal. `None` until the first turn ends.
    cost_usage: Option<hive_claude::TokenUsage>,
    /// `needs_login`, from this process's own [`LoginState`], so always set.
    #[serde(flatten)]
    login: Login,
}

/// What this process can honestly report about the agent-state axis.
///
/// Only two of [`AgentState`]'s four variants are knowable from inside the
/// container, and the split is not arbitrary:
///
/// - [`AgentState::Up`] and [`AgentState::Paused`] are the agent's own
///   facts. Running is self-evident — this code is executing — and the
///   pause marker is a file in the harness dir that this process reads on
///   every turn-loop iteration anyway (`crate::paths::paused_marker`).
/// - [`AgentState::Offline`] and [`AgentState::Destroyed`] are **not
///   reportable from here, ever**. A stopped agent publishes nothing and a
///   destroyed one does not exist, so a message claiming either would be a
///   message from a process contradicting itself. Those two remain
///   hive-c0re's to observe and the `agent-status` bucket's to carry; a
///   reader wanting the full four-state picture needs both sources, and
///   this one's silence is the only evidence it can offer for the other
///   two.
///
/// Read fresh per publish rather than cached: the marker is written by
/// hive-c0re (through hive-priv) and by this process's own auto-pause, so
/// nothing in here observes every write.
fn observed_state() -> AgentState {
    if crate::paths::paused_marker().exists() {
        AgentState::Paused
    } else {
        AgentState::Up
    }
}

/// The header's login fields for `state`.
fn login(state: LoginState) -> Login {
    Login {
        needs_login: Some(state == LoginState::NeedsLogin),
    }
}

/// Build the current header from the bus, the login state and the pause
/// marker.
fn snapshot(bus: &Bus, login_state: &Mutex<LoginState>) -> AgentStateMsg {
    let (turn_state, since_unix) = bus.state_snapshot();
    let model = bus.model();
    AgentStateMsg {
        turn_state,
        turn_state_since: iso8601_utc(since_unix),
        agent_state: observed_state(),
        context_window_tokens: bus.effective_context_window(&model),
        inference: Effective {
            model,
            effort: bus.effort(),
            model_source: bus.model_source(),
            effort_source: bus.effort_source(),
        },
        resolved_model: bus.last_resolved_model(),
        ctx_usage: bus.last_ctx_usage(),
        cost_usage: bus.last_cost_usage(),
        login: login(*login_state.lock().unwrap_or_else(PoisonError::into_inner)),
    }
}

/// Start the publish task, if this agent has a label to name its subject with.
///
/// Every writer of `login_state` emits a status on `bus` after the write, so
/// a login change goes out as an ordinary rebuild.
pub fn spawn(bus: &Bus, login_state: Arc<Mutex<LoginState>>) {
    let agent = crate::identity::label();
    if agent.is_empty() {
        tracing::warn!("this agent has no label; not publishing turn state upward");
        return;
    }
    tokio::spawn(run(bus.subscribe(), bus.clone(), login_state, agent));
}

/// Watch the bus and publish whenever the header actually changed.
///
/// **Why the change test is on the serialized bytes rather than on which
/// event arrived.** The header is assembled from independent pieces of
/// `Bus` state, only some of which announce themselves with an event of
/// their own — `set_resolved_model`, for one, emits nothing and lands
/// alongside the `TokenUsageChanged` of the same turn. An allow-list of
/// interesting variants would therefore have to encode which event happens
/// to be adjacent to each field's write, and would go quietly stale the
/// first time a field moved. Rebuilding on (almost) any event and comparing
/// the result is the same publish traffic with none of that coupling: a
/// rebuild is a handful of mutex reads and one `stat`.
///
/// `Stream` is the exception, skipped before the rebuild: it is the only
/// high-rate variant — one per `stream-json` line, so hundreds per turn —
/// and it carries nothing this header reads. Every other variant, including
/// any added later, funnels into the comparison and costs nothing when it
/// changes nothing.
///
/// A [`Trigger::Refresh`] re-sends the header even when it is unchanged.
async fn run(
    mut rx: broadcast::Receiver<BusEvent>,
    bus: Bus,
    login_state: Arc<Mutex<LoginState>>,
    agent: String,
) {
    let client = crate::swarm_queue::client().await;
    let subject = swarm_queue_client::agent_state_subject(&agent);
    tracing::info!(subject, "publishing agent turn state to the swarm queue");

    // The last payload actually sent, so a rebuild that changed nothing is
    // dropped here instead of on the wire. `None` until the first publish.
    let mut last: Option<Vec<u8>> = None;
    let mut refresh = refresh_timer();
    while let Some(trigger) = next_trigger(&mut rx, &mut refresh).await {
        let msg = snapshot(&bus, &login_state);
        publish(&client, &subject, &msg, &trigger, &mut last).await;
    }
}

/// Whether `payload` goes out: on a refresh always, and `last` is forgotten
/// so a refresh that fails is re-sent by the next event; otherwise only when
/// it differs from `last`.
fn due(trigger: &Trigger, last: &mut Option<Vec<u8>>, payload: &[u8]) -> bool {
    if *trigger == Trigger::Refresh {
        *last = None;
    }
    last.as_deref() != Some(payload)
}

/// Why a header is offered.
#[derive(Debug, PartialEq, Eq)]
enum Trigger {
    /// A bus event that may have changed it.
    Event,
    /// [`REFRESH_INTERVAL`] passed.
    Refresh,
}

/// Ticks every [`REFRESH_INTERVAL`], the first one an interval from now,
/// since the harness's boot events already send the first header.
fn refresh_timer() -> Interval {
    let mut timer = tokio::time::interval_at(
        tokio::time::Instant::now() + REFRESH_INTERVAL,
        REFRESH_INTERVAL,
    );
    timer.set_missed_tick_behavior(MissedTickBehavior::Delay);
    timer
}

/// The next reason to offer a header; `None` once the bus is closed.
async fn next_trigger(
    rx: &mut broadcast::Receiver<BusEvent>,
    refresh: &mut Interval,
) -> Option<Trigger> {
    loop {
        tokio::select! {
            _ = refresh.tick() => return Some(Trigger::Refresh),
            event = rx.recv() => match event {
                // Skipped before the rebuild — see [`run`]'s doc.
                Ok(BusEvent {
                    event: LiveEvent::Stream(_),
                    ..
                }) => {}
                Ok(_) => return Some(Trigger::Event),
                // The bus drops events for a subscriber that falls behind. A
                // header is a current value rather than a log, so a missed
                // event costs nothing here: the rebuild reads the state that
                // the missed events led to, not the events themselves.
                Err(broadcast::error::RecvError::Lagged(missed)) => {
                    tracing::debug!(missed, "swarm agent state: lagged, rebuilding anyway");
                    return Some(Trigger::Event);
                }
                // Every sender is gone, so the harness is shutting down.
                Err(broadcast::error::RecvError::Closed) => return None,
            },
        }
    }
}

/// Offer one header, if [`due`]. Every failure is terminal for that publish
/// and for nothing else.
///
/// `last` is only updated on a publish the client accepted, so a transition
/// lost to a transport failure is re-sent by the next one rather than
/// deduped away against a value the swarm never saw.
async fn publish(
    client: &async_nats::Client,
    subject: &str,
    msg: &AgentStateMsg,
    trigger: &Trigger,
    last: &mut Option<Vec<u8>>,
) {
    let payload = match serde_json::to_vec(msg) {
        Ok(payload) => payload,
        Err(e) => {
            tracing::warn!(error = %e, "swarm agent state: serialising failed, header dropped");
            return;
        }
    };
    if !due(trigger, last, &payload) {
        return;
    }
    // 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.
    if let Err(e) = swarm_queue_client::ensure_connected(client) {
        tracing::warn!(error = %swarm_queue_client::chain(&e), "swarm agent state: publish skipped");
        return;
    }
    // No degrade path, unlike `swarm_term`'s: every field here is a number,
    // an enum, or a model name, so there is no arbitrary-length body worth
    // spending to get under the limit. The check stays because being one
    // byte over does not truncate the message, it gets it refused and the
    // connection closed — which would cost every header racing behind it
    // through the reconnect, not just this one.
    let limit = swarm_queue_client::max_payload(client).saturating_sub(HEADROOM);
    if payload.len() > limit {
        tracing::warn!(
            len = payload.len(),
            limit,
            "swarm agent state: header over the payload limit, dropped"
        );
        return;
    }
    if let Err(e) = client
        .publish(subject.to_owned(), payload.clone().into())
        .await
    {
        tracing::warn!(error = %e, "swarm agent state: publish failed, header dropped");
        return;
    }
    *last = Some(payload);
}

#[cfg(test)]
mod tests {
    use super::{
        AgentStateMsg, REFRESH_INTERVAL, Trigger, due, login, next_trigger, refresh_timer, snapshot,
    };
    use crate::events::{BusEvent, LiveEvent, TurnState};
    use crate::login::LoginState;
    use std::sync::Mutex;
    use swarm_wire::agent_inference::{Effective, Source};
    use swarm_wire::wanted::AgentState;
    use tokio::sync::broadcast;
    use tokio::time::{Duration, Instant};

    /// `TokenUsage` is `#[non_exhaustive]`, so a downstream crate cannot
    /// write one as a struct expression at all — see `turn_stats`'s note on
    /// the same restriction. Deserializing one is the shortest way to a
    /// populated fixture here, and it keeps the numbers next to the field
    /// names they belong to.
    ///
    /// ⚠️ All four fields, every time: none of them carries a serde default,
    /// so a block written with only the field a given test reads fails to
    /// deserialize at all rather than zero-filling the rest.
    fn usage(json: &serde_json::Value) -> hive_claude::TokenUsage {
        serde_json::from_value(json.clone()).expect("a usage block")
    }

    /// A header with every optional field populated, so a test asserting on
    /// the serialized shape sees the widest form a reader can receive.
    fn msg() -> AgentStateMsg {
        AgentStateMsg {
            turn_state: TurnState::Thinking,
            // As `iso8601_utc` renders a transition time.
            turn_state_since: "2026-09-13T12:35:03Z".to_owned(),
            agent_state: AgentState::Paused,
            inference: Effective {
                model: "sonnet".to_owned(),
                effort: "high".to_owned(),
                model_source: Source::Swarm,
                effort_source: Source::AgentNix,
            },
            resolved_model: Some("claude-sonnet-4-5-20260805".to_owned()),
            context_window_tokens: 200_000,
            ctx_usage: Some(usage(&serde_json::json!({
                "input_tokens": 11,
                "output_tokens": 22,
                "cache_read_input_tokens": 33,
                "cache_creation_input_tokens": 44,
            }))),
            // Distinct from `ctx_usage`'s numbers in every field, so a
            // renderer served one block where it asked for the other shows
            // up as a wrong value rather than as a coincidence.
            cost_usage: Some(usage(&serde_json::json!({
                "input_tokens": 55,
                "output_tokens": 66,
                "cache_read_input_tokens": 77,
                "cache_creation_input_tokens": 88,
            }))),
            login: login(LoginState::NeedsLogin),
        }
    }

    /// The published contract, asserted on the JSON a subscriber parses
    /// rather than on the Rust struct: a renderer is written against these
    /// key names, and renaming a field in Rust without noticing is exactly
    /// the failure this catches.
    #[test]
    fn the_payload_carries_the_contracts_field_names() {
        let value: serde_json::Value =
            serde_json::from_slice(&serde_json::to_vec(&msg()).expect("serialises"))
                .expect("valid json");
        let obj = value.as_object().expect("an object");
        let mut keys: Vec<&str> = obj.keys().map(String::as_str).collect();
        keys.sort_unstable();
        assert_eq!(
            keys,
            [
                "agent_state",
                "context_window_tokens",
                "cost_usage",
                "ctx_usage",
                "effort",
                "effort_source",
                "model",
                "model_source",
                "needs_login",
                "resolved_model",
                "turn_state",
                "turn_state_since",
            ]
        );
    }

    /// swarm-controller reads the inference fields back out of a header.
    #[test]
    fn the_header_decodes_as_its_inference_settings() {
        let bytes = serde_json::to_vec(&msg()).expect("serialises");
        let decoded: Effective = serde_json::from_slice(&bytes).expect("decodes");
        assert_eq!(decoded, msg().inference);
    }

    /// Both enums cross the wire as the snake-case strings their `serde`
    /// attributes promise, not as Rust variant names — and the two stay
    /// separate fields, since neither vocabulary contains the other's values.
    #[test]
    fn the_two_state_axes_serialise_as_their_own_snake_case_strings() {
        let value = serde_json::to_value(msg()).expect("serialises");
        assert_eq!(value["turn_state"], serde_json::json!("thinking"));
        assert_eq!(value["agent_state"], serde_json::json!("paused"));
    }

    /// The departure from `StateSnapshot` that a reader is most likely to
    /// get wrong: this field is a string, and a number here would parse as a
    /// valid — and completely wrong — date on the other side.
    #[test]
    fn turn_state_since_is_an_iso_8601_string_not_unix_seconds() {
        let value = serde_json::to_value(msg()).expect("serialises");
        assert_eq!(
            value["turn_state_since"],
            serde_json::json!("2026-09-13T12:35:03Z")
        );
    }

    /// The token blocks ship whole rather than pre-summed: a header renderer
    /// wanting the context percentage needs the same three fields
    /// `TokenUsage::context_tokens` adds up, and one of them alone reads as
    /// near-zero once prompt caching is on.
    #[test]
    fn the_usage_blocks_ship_their_own_fields() {
        let value = serde_json::to_value(msg()).expect("serialises");
        assert_eq!(value["ctx_usage"]["cache_read_input_tokens"], 33);
        assert_eq!(value["cost_usage"]["input_tokens"], 55);
        assert_eq!(value["context_window_tokens"], 200_000);
    }

    /// A pre-first-turn header, which is what a freshly booted agent
    /// actually publishes. The nullable fields have to be present and null
    /// rather than absent — a renderer that reads `ctx_usage` off a header
    /// missing the key gets `undefined`, which is not the same thing as
    /// "this agent has not finished a turn yet".
    #[test]
    fn a_header_from_before_the_first_turn_still_carries_every_key() {
        let fresh = AgentStateMsg {
            turn_state: TurnState::Idle,
            agent_state: AgentState::Up,
            resolved_model: None,
            ctx_usage: None,
            cost_usage: None,
            ..msg()
        };
        let value = serde_json::to_value(fresh).expect("serialises");
        assert_eq!(value["resolved_model"], serde_json::Value::Null);
        assert_eq!(value["ctx_usage"], serde_json::Value::Null);
        assert_eq!(value["cost_usage"], serde_json::Value::Null);
        assert_eq!(value["turn_state"], serde_json::json!("idle"));
        assert_eq!(value["agent_state"], serde_json::json!("up"));
    }

    /// swarm-controller reads the login flag back out of a header.
    #[test]
    fn the_header_decodes_as_its_login_flag() {
        for state in [LoginState::NeedsLogin, LoginState::Online] {
            let bytes = serde_json::to_vec(&AgentStateMsg {
                login: login(state),
                ..msg()
            })
            .expect("serialises");
            let decoded: swarm_wire::agent_login::Login =
                serde_json::from_slice(&bytes).expect("decodes");
            assert_eq!(
                decoded.needs_login,
                Some(state == LoginState::NeedsLogin),
                "{state:?}"
            );
        }
    }

    /// A login landing, or a logout, changes the header the next bus event
    /// rebuilds, so it goes out without waiting for a refresh.
    #[test]
    fn a_login_state_change_is_published_on_the_next_event() {
        let (before, after) = crate::paths::with_harness_dir(|| {
            let bus = crate::events::Bus::new();
            let login_state = Mutex::new(LoginState::NeedsLogin);
            let before = serde_json::to_vec(&snapshot(&bus, &login_state)).expect("serialises");
            *login_state.lock().unwrap() = LoginState::Online;
            let after = serde_json::to_vec(&snapshot(&bus, &login_state)).expect("serialises");
            (before, after)
        });
        let mut last = Some(before.clone());
        assert!(!due(&Trigger::Event, &mut last, &before));
        assert!(due(&Trigger::Event, &mut last, &after));
        let after: serde_json::Value = serde_json::from_slice(&after).expect("valid json");
        assert_eq!(after["needs_login"], serde_json::json!(false));
    }

    fn event(event: LiveEvent) -> BusEvent {
        BusEvent {
            seq: 0,
            ts: 0,
            event,
        }
    }

    /// An idle bus still yields a refresh once per interval; a bus event
    /// yields at once, a stream line alone yields nothing, and a closed bus
    /// ends the loop.
    #[tokio::test(start_paused = true)]
    async fn an_idle_header_is_refreshed_once_per_interval() {
        let (tx, mut rx) = broadcast::channel(16);
        let mut refresh = refresh_timer();
        let start = Instant::now();
        assert_eq!(
            next_trigger(&mut rx, &mut refresh).await,
            Some(Trigger::Refresh)
        );
        assert_eq!(start.elapsed(), REFRESH_INTERVAL);

        tx.send(event(LiveEvent::Stream(serde_json::json!({}))))
            .unwrap();
        tx.send(event(LiveEvent::Note {
            text: "claude exited 0".to_owned(),
        }))
        .unwrap();
        let at = Instant::now();
        assert_eq!(
            next_trigger(&mut rx, &mut refresh).await,
            Some(Trigger::Event)
        );
        assert_eq!(at.elapsed(), Duration::ZERO);

        assert_eq!(
            next_trigger(&mut rx, &mut refresh).await,
            Some(Trigger::Refresh)
        );
        assert_eq!(start.elapsed(), 2 * REFRESH_INTERVAL);

        drop(tx);
        assert_eq!(next_trigger(&mut rx, &mut refresh).await, None);
    }

    #[test]
    fn an_unchanged_header_goes_out_on_a_refresh_only() {
        let sent = b"{\"turn_state\":\"idle\"}";
        let mut last = Some(sent.to_vec());
        assert!(!due(&Trigger::Event, &mut last, sent));
        assert!(due(&Trigger::Refresh, &mut last, sent));
    }

    #[test]
    fn a_changed_header_goes_out_on_either_trigger() {
        let changed = b"{\"turn_state\":\"thinking\"}";
        for trigger in [Trigger::Event, Trigger::Refresh] {
            let mut last = Some(b"{\"turn_state\":\"idle\"}".to_vec());
            assert!(due(&trigger, &mut last, changed), "{trigger:?}");
        }
        // Nothing sent yet.
        assert!(due(&Trigger::Event, &mut None, changed));
    }

    /// A refresh the client then fails to send leaves nothing to dedup the
    /// next event against, so that event re-sends the header.
    #[test]
    fn a_failed_refresh_is_re_sent_by_the_next_event() {
        let sent = b"{\"turn_state\":\"idle\"}";
        let mut last = Some(sent.to_vec());
        assert!(due(&Trigger::Refresh, &mut last, sent));
        // `publish` records `last` only once the client accepts the payload.
        assert!(due(&Trigger::Event, &mut last, sent));
    }
}