hyperhive/hive-c0re/src/workers/wanted.rs

//! Converging this hive onto the agent set the swarm controller declares.
//!
//! The controller gives each hive its own `hive-wanted-<hive>` bucket (see
//! [`swarm_wire::wanted`]); this reads its own and acts on it. It is
//! the **repair** path, not the fast one: a deploy event
//! ([`crate::swarm_status`]) is core NATS, so a hive that was down never hears
//! it, and until this existed that left the hive on its old shape with nothing
//! looking wrong.
//!
//! # Two things a declaration does not mean
//!
//! **Absence is not a deletion order.** An agent missing from the value — or
//! no value, or no bucket — is one the controller has said nothing about.
//! Swarm-side lifecycle does not yet cover agents that predate it, so
//! converging "to exactly this set" would tear down every agent the swarm has
//! not adopted.
//!
//! **An unknown state is not a partial instruction.** [`AgentState`] is closed,
//! so a value this build cannot read fails the whole decode: the hive converges
//! nothing rather than obeying the agents it happened to understand.
//!
//! # What "diverged" is measured against
//!
//! The hive's **stored power intent**, not the container's observed running
//! state. An agent down while its intent says `Up` is already the boot
//! reconcile's work; a loop reading `is_running` would insert a start DAG
//! behind that reconcile's back on every boot.
//!
//! One consequence, deliberate: a declaration outranks a local stop — an agent
//! declared `Up` returns at the next boot or watch attach, because the controller owns that.

use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;
use std::time::Duration;

use anyhow::{Context, Result};
use swarm_wire::wanted::{AgentState, HiveWanted};

use crate::coordinator::Coordinator;
use crate::power::Wanted;

/// Longest `pull`, and each watch session's attach and first pull together, wait
/// for the swarm queue to connect.
const CONNECT_WAIT: Duration = Duration::from_mins(1);

/// How often the wait for a connected client looks at the connection state.
const CONNECT_POLL: Duration = Duration::from_millis(250);

/// Wait after the first failed watch session; doubled after each further one.
const FIRST_RETRY_DELAY: Duration = Duration::from_secs(5);

/// The longest wait between two watch sessions.
const MAX_RETRY_DELAY: Duration = Duration::from_mins(5);

/// Read this hive's declaration and converge the agents it names.
///
/// `Ok(())` covers three quiet cases that are **not** an empty declaration: no
/// hive name, no bucket, no key. Each logs its own line, because a loop that
/// silently does nothing and one that is working look identical otherwise.
/// Anything else — an unreachable queue, an undecodable value, an agent list
/// this hive cannot enumerate — is an error, so the node fails visibly on the
/// dashboard instead of reading as "nothing was declared".
///
/// Waits up to [`CONNECT_WAIT`] in total, for the client and then for its
/// connection: at boot the shared client exists before it is connected, and a
/// pull that did not wait would fail every boot.
///
/// # Errors
/// Propagates a failed queue connect, KV read, decode, or agent enumeration.
pub async fn pull(coord: &Arc<Coordinator>) -> Result<()> {
    let Some(hive) = crate::container_view::hive_swarm_names().0 else {
        tracing::debug!("wanted state: HYPERHIVE_HIVE_NAME unset; no key to read");
        return Ok(());
    };
    let deadline = tokio::time::Instant::now() + CONNECT_WAIT;
    let client =
        match tokio::time::timeout_at(deadline, crate::swarm_queue::wait_for_client()).await {
            // Absent — already reported by the shared connector (an `info` log or
            // a `swarm_queue_config` warning).
            Ok(None) => return Ok(()),
            Ok(Some(client)) => client,
            Err(_) => anyhow::bail!("no swarm queue client within {}s", CONNECT_WAIT.as_secs()),
        };
    pull_with(coord, &client, &hive, deadline).await
}

/// [`pull`] over a client the caller already has, waiting for its connection
/// no later than `deadline`.
async fn pull_with(
    coord: &Arc<Coordinator>,
    client: &async_nats::Client,
    hive: &str,
    deadline: tokio::time::Instant,
) -> Result<()> {
    wait_connected(client, deadline).await?;

    let Some(store) = swarm_queue_client::wanted::open_read_only(client, hive)
        .await
        .context("opening this hive's wanted-state bucket")?
    else {
        tracing::info!(%hive, "wanted state: no bucket yet; nothing has been declared");
        return Ok(());
    };
    let Some(raw) = store
        .get(hive)
        .await
        .context("reading this hive's wanted state")?
    else {
        tracing::info!(%hive, "wanted state: no key for this hive; nothing has been declared");
        return Ok(());
    };
    let declared: HiveWanted =
        serde_json::from_slice(&raw).context("decoding this hive's wanted state")?;

    converge(coord, &declared).await
}

/// Wait until `deadline` for `client` to be connected.
///
/// `swarm_queue::wait_for_client` returns a client that may still be
/// connecting. An unconnected client does not fail a `JetStream` request, it
/// hangs on it, and a hung node shows as running on the dashboard, so every
/// request here is preceded by this bounded wait and its
/// `ensure_connected` verdict.
async fn wait_connected(client: &async_nats::Client, deadline: tokio::time::Instant) -> Result<()> {
    while swarm_queue_client::ensure_connected(client).is_err()
        && tokio::time::Instant::now() < deadline
    {
        tokio::time::sleep(CONNECT_POLL).await;
    }
    Ok(swarm_queue_client::ensure_connected(client)?)
}

/// How one watch session ended.
#[derive(Debug, PartialEq, Eq)]
enum SessionEnd {
    /// The hive is shutting down.
    Shutdown,
    /// The watch could not be attached (`attached == false`) or its stream
    /// ended.
    Lost { attached: bool },
}

/// The wait before the next watch session after `failures` sessions in a row
/// never attached: [`FIRST_RETRY_DELAY`] doubling up to [`MAX_RETRY_DELAY`].
fn retry_delay(failures: u32) -> Duration {
    FIRST_RETRY_DELAY
        .saturating_mul(1_u32 << failures.min(16))
        .min(MAX_RETRY_DELAY)
}

/// Attach the watch, then pull once, and hand back the watch.
///
/// The watch comes first: a put landing between a pull and a later attach
/// would be seen by neither. One landing between the two is seen by both, which
/// `converge` tolerates. `None` when the attach did not happen; a failed pull is
/// logged and does not drop the watch, since the next attach pulls again.
async fn attach_then_pull<U, A, P>(attach: A, pull: P) -> Option<U>
where
    A: Future<Output = Option<U>>,
    P: Future<Output = Result<()>>,
{
    let updates = attach.await?;
    if let Err(e) = pull.await {
        tracing::warn!(error = ?e, "wanted state: pull after attaching the watch failed");
    }
    Some(updates)
}

/// Run watch sessions until shutdown, waiting [`retry_delay`] between them.
async fn supervise<F, Fut>(mut session: F, mut shutdown: tokio::sync::watch::Receiver<bool>)
where
    F: FnMut() -> Fut,
    Fut: Future<Output = SessionEnd>,
{
    let mut failures = 0_u32;
    loop {
        match session().await {
            SessionEnd::Shutdown => return,
            SessionEnd::Lost { attached } => {
                failures = if attached {
                    0
                } else {
                    failures.saturating_add(1)
                };
            }
        }
        tokio::select! {
            () = tokio::time::sleep(retry_delay(failures)) => {}
            _ = shutdown.changed() => return,
        }
    }
}

/// Converge again each time the controller republishes this hive's declaration.
///
/// The **fast** path. Every attach, the first included, is followed by one
/// [`pull`]: the watch hears only what is published while it is attached, so
/// what was declared before an attach (a hive that was down, a deploy lost to
/// a restart) would otherwise wait for the next boot. A failed attach or an
/// ended stream is retried with a bounded backoff, forever.
///
/// ⚠️ A refused watch and a quiet one are told apart here, unlike the core-NATS
/// subscriptions in [`crate::swarm_status`]: a watch is a `JetStream` consumer,
/// so the request is answered, and a hive lacking the grant gets an error rather
/// than silence. That is why this warns instead of returning quietly.
pub async fn watch_declarations(
    client: async_nats::Client,
    coord: Arc<Coordinator>,
    hive: String,
    shutdown: tokio::sync::watch::Receiver<bool>,
) {
    Box::pin(supervise(
        || watch_session(&client, &coord, &hive, shutdown.clone()),
        shutdown.clone(),
    ))
    .await;
}

/// One attach, one pull, then the updates until the stream ends.
async fn watch_session(
    client: &async_nats::Client,
    coord: &Arc<Coordinator>,
    hive: &str,
    mut shutdown: tokio::sync::watch::Receiver<bool>,
) -> SessionEnd {
    use futures_util::StreamExt as _;

    let deadline = tokio::time::Instant::now() + CONNECT_WAIT;
    let attach = async {
        if let Err(e) = wait_connected(client, deadline).await {
            tracing::warn!(%hive, error = ?e, "wanted state: cannot watch this hive's bucket; retrying");
            return None;
        }
        match swarm_queue_client::wanted::watch(client, hive).await {
            Ok(Some(updates)) => Some(updates),
            Ok(None) => {
                tracing::warn!(%hive, "wanted state: cannot watch this hive's bucket; retrying");
                None
            }
            Err(e) => {
                tracing::warn!(
                    %hive,
                    error = %swarm_queue_client::chain(&e),
                    "wanted state: cannot watch this hive's bucket; retrying"
                );
                None
            }
        }
    };
    let Some(mut updates) = Box::pin(attach_then_pull(
        attach,
        pull_with(coord, client, hive, deadline),
    ))
    .await
    else {
        return SessionEnd::Lost { attached: false };
    };
    tracing::info!(%hive, "wanted state: watching for declarations");

    loop {
        tokio::select! {
            entry = updates.next() => {
                let Some(entry) = entry else {
                    // `async-nats` reconnects underneath a live watch, so an
                    // ended stream is the connection going away for good
                    // rather than a blip to spin on.
                    tracing::warn!(%hive, "wanted state: watch closed; re-attaching");
                    return SessionEnd::Lost { attached: true };
                };
                match entry {
                    Ok(entry) => apply_entry(coord, hive, &entry).await,
                    // The watch survives one bad entry; the stream ending is
                    // the case above.
                    Err(e) => tracing::warn!(%hive, error = %e, "wanted state: watch error"),
                }
            }
            _ = shutdown.changed() => {
                tracing::info!("wanted state: shutdown signal received");
                return SessionEnd::Shutdown;
            }
        }
    }
}

/// Converge one watched update, or decline to.
///
/// A delete is **not** a deletion order — the module docs' rule, and the reason
/// this is not simply "decode and converge": the controller removing the key
/// says nothing about the agents this hive runs, so acting on it would tear
/// down the very set that absence is defined not to touch.
async fn apply_entry(
    coord: &Arc<Coordinator>,
    hive: &str,
    entry: &async_nats::jetstream::kv::Entry,
) {
    if !carries_a_declaration(entry.operation) {
        tracing::info!(%hive, "wanted state: declaration withdrawn; nothing to converge");
        return;
    }
    let declared: HiveWanted = match serde_json::from_slice(&entry.value) {
        Ok(declared) => declared,
        // Warned rather than propagated: this end and the controller share one
        // type, so a decode failure means they disagree about it — and the
        // watch must keep running to pick up the next, possibly good, value.
        Err(e) => {
            tracing::warn!(%hive, error = %e, "wanted state: undecodable declaration");
            return;
        }
    };
    if let Err(e) = converge(coord, &declared).await {
        tracing::warn!(%hive, error = ?e, "wanted state: converging a watched update failed");
    }
}

/// Whether a watched operation carries a declaration to converge to.
///
/// Pure, and separate from [`apply_entry`], so the module's "absence is not a
/// deletion order" rule is enforced by a test rather than only asserted in
/// prose — converging on a removed key is the one mistake here that would tear
/// down agents nobody asked to stop.
fn carries_a_declaration(operation: async_nats::jetstream::kv::Operation) -> bool {
    use async_nats::jetstream::kv::Operation;

    match operation {
        Operation::Put => true,
        Operation::Delete | Operation::Purge => false,
    }
}

/// Queue whatever the declaration asks for and this hive is not already doing.
async fn converge(coord: &Arc<Coordinator>, declared: &HiveWanted) -> Result<()> {
    // Fail closed, for the reason the deploy event's own arm gives: without
    // the list, "this hive does not have that agent" cannot be told from
    // "this hive cannot see its agents", and acting on the guess creates a
    // container over one that already exists.
    let present: BTreeSet<String> = crate::lifecycle::agents_for_meta_listing()
        .await
        .context("enumerating this hive's agents")?
        .into_iter()
        .map(|spec| spec.name)
        .collect();

    let mut intents = BTreeMap::new();
    for agent in declared.agents.keys() {
        match coord.power.get(agent) {
            Ok(intent) => {
                intents.insert(agent.clone(), intent);
            }
            // Left out of the map, which is what makes `plan` skip it.
            Err(e) => {
                tracing::warn!(%agent, error = ?e, "wanted state: cannot read power intent; skipping");
            }
        }
    }

    // The pause marker is a stat, not a store read — it cannot fail the way
    // `coord.power.get` above can, so there is no "unreadable, skip" branch
    // to mirror. A name that does not parse as an `Ident` is skipped with a
    // warning instead, the same defensive shape `crash_watch` uses for the
    // same reason: a declaration is attacker-adjacent input (published by
    // the swarm controller, not typed by an operator at this hive), so a
    // malformed name should not panic the convergence loop.
    let mut paused = BTreeMap::new();
    for agent in declared.agents.keys() {
        match hive_types::Ident::parse(agent) {
            Ok(id) => {
                paused.insert(agent.clone(), Coordinator::is_paused(&id));
            }
            Err(e) => {
                tracing::warn!(%agent, error = %e, "wanted state: invalid agent name; skipping pause check");
            }
        }
    }

    let plan = plan(declared, &present, &intents, &paused);
    tracing::info!(
        declared = declared.agents.len(),
        deploy = plan.deploy.len(),
        start = plan.start.len(),
        stop = plan.stop.len(),
        destroy = plan.destroy.len(),
        pause = plan.pause.len(),
        resume = plan.resume.len(),
        "wanted state: converging"
    );

    let mut queued = false;
    for agent in &plan.deploy {
        match crate::swarm_status::queue_first_deploy(coord, agent) {
            Ok(Some(_)) => queued = true,
            // Already queued or running: the pull and the watch overlap.
            Ok(None) => {}
            // Per agent rather than the whole sweep: one agent that cannot be
            // queued is not a reason to leave the others undeclared.
            Err(e) => tracing::warn!(%agent, error = %e, "wanted state: queueing a deploy failed"),
        }
    }
    if !plan.start.is_empty() {
        crate::job_queue::power::start_many(coord, &plan.start)
            .await
            .context("queueing the declared starts")?;
    }
    if !plan.stop.is_empty() {
        crate::job_queue::power::stop_many(coord, &plan.stop, true)
            .await
            .context("queueing the declared stops")?;
    }
    // No purge: this issue's own scope is tearing the container down, not
    // wiping its persistent state. `actions::destroy` submits its own DAG
    // node and emits its own queue snapshot, so there is nothing to await
    // or fold into `queued` here — same fire-and-forget contract every other
    // lifecycle op in that module has.
    for agent in &plan.destroy {
        crate::actions::destroy(coord, agent, false);
    }
    if queued {
        // The power ops and destroy emit their own; the first deploys do not.
        coord.emit_rebuild_queue_snapshot();
    }

    // `set_paused_by_name` — same helper `dashboard::lifecycle_ops`'s
    // `/api/pause`/`/api/resume` use, not the job-queue DAG those
    // name-check in their own doc comments — this loop already has its own
    // per-agent retry story (redeclare-or-rewatch), so there is nothing
    // here to wait on an acknowledgement for. One loop over both lists
    // paired with their target `paused` value, not two near-identical
    // copies (argus, PR review).
    let mut rescan = false;
    for (agents, paused) in [(&plan.pause, true), (&plan.resume, false)] {
        for agent in agents {
            match Coordinator::set_paused_by_name(agent, paused).await {
                Ok(()) => rescan = true,
                Err(e) => {
                    tracing::warn!(%agent, paused, error = %e, "wanted state: set-paused failed");
                }
            }
        }
    }
    if rescan {
        // Same call `post_pause`/`post_resume` make, so the dashboard's
        // "paused" badge flips without waiting for the next periodic sweep.
        coord.rescan_containers_and_emit().await;
    }
    Ok(())
}

/// Everything one declaration asks this hive to queue.
///
/// `pause`/`resume` are a second, independent axis from the other four —
/// see [`decide_pause`]'s own comment for why an agent can land in one of
/// the power-axis lists *and* one of these in the same pass.
#[derive(Debug, Default, PartialEq, Eq)]
struct Plan {
    deploy: Vec<String>,
    start: Vec<String>,
    stop: Vec<String>,
    destroy: Vec<String>,
    pause: Vec<String>,
    resume: Vec<String>,
}

/// Turn one declaration into that plan — pure, so what the loop does with a
/// given declaration is testable without a queue, a container or a store.
///
/// `intents` is the hive's stored power intent per agent, and an agent
/// **missing from the map** is one whose intent could not be read: it is
/// skipped. That is not `Some(None)` — "no row yet" is a reading and gets
/// converged, "the read failed" is not and would be a guess.
///
/// An agent the declaration does not name reaches none of the three lists.
/// That is the whole of "absence is not a deletion order": the agents this
/// hive runs that the swarm has not adopted are never even inspected.
fn plan(
    declared: &HiveWanted,
    present: &BTreeSet<String>,
    intents: &BTreeMap<String, Option<Wanted>>,
    paused: &BTreeMap<String, bool>,
) -> Plan {
    let mut plan = Plan::default();
    for (agent, decl) in &declared.agents {
        // The pause axis does not gate on a readable power intent the way
        // the match below does — it has no intent row to fail reading, only
        // a stat that always answers — so it is evaluated unconditionally,
        // even for an agent this pass otherwise skips.
        let is_present = present.contains(agent);
        let is_paused = paused.get(agent).copied().unwrap_or(false);
        match decide_pause(decl.state, is_present, is_paused) {
            PauseConverge::Pause => plan.pause.push(agent.clone()),
            PauseConverge::Resume => plan.resume.push(agent.clone()),
            PauseConverge::Nothing => {}
        }

        let Some(intent) = intents.get(agent) else {
            continue;
        };
        match decide(decl.state, is_present, *intent) {
            Converge::Deploy => plan.deploy.push(agent.clone()),
            Converge::Start => plan.start.push(agent.clone()),
            Converge::Stop => plan.stop.push(agent.clone()),
            Converge::Destroy => plan.destroy.push(agent.clone()),
            Converge::Nothing => {}
        }
    }
    plan
}

/// What the loop will do about one declared agent.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Converge {
    /// Declared, and this hive has no container for it — the missed-deploy
    /// repair.
    Deploy,
    Start,
    Stop,
    /// Declared destroyed, and this hive still has a container for it — run
    /// the teardown once. Irreversible, so unlike `Start`/`Stop` there is no
    /// intent to compare against: presence alone decides it.
    Destroy,
    /// The hive already agrees with the declaration.
    Nothing,
}

/// The whole decision, pure: no queue, no store, no container. The three
/// inputs are exactly what [`converge`] reads per agent, so the table below is
/// the behaviour rather than a model of it.
///
/// `Paused` shares every one of `Up`'s arms here — on the power axis a
/// paused agent is still a running one, just with its turn loop parked.
/// [`decide_pause`] is where the two diverge.
fn decide(state: AgentState, present: bool, intent: Option<Wanted>) -> Converge {
    match state {
        AgentState::Up | AgentState::Paused if !present => Converge::Deploy,
        AgentState::Up | AgentState::Paused => {
            if intent == Some(Wanted::Up) {
                Converge::Nothing
            } else {
                Converge::Start
            }
        }
        // Declared offline/destroyed and not here: creating a container in
        // order to leave it stopped, or to immediately re-destroy it, is not
        // what either declaration asks for — absence plus a stopped-or-gone
        // declaration is agreement, not a repair to queue.
        AgentState::Offline | AgentState::Destroyed if !present => Converge::Nothing,
        AgentState::Offline => {
            if intent == Some(Wanted::Offline) {
                Converge::Nothing
            } else {
                Converge::Stop
            }
        }
        // Power intent plays no part here — a destroyed agent has no
        // running/stopped distinction left to converge to, only present or
        // not.
        AgentState::Destroyed => Converge::Destroy,
    }
}

/// What the loop will do about one declared agent's turn-loop pause state.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PauseConverge {
    Pause,
    Resume,
    /// Either the pause marker already matches the declaration, or there is
    /// no container yet to hold one — see this function's own comment on
    /// `!present`.
    Nothing,
}

/// The pause-axis half of [`decide`], deliberately **not** folded into it:
/// the two vary independently (an agent can need `Converge::Start` and
/// `PauseConverge::Pause` in the same pass — declared `Paused`, currently
/// stopped), so a single combined enum would need one variant per
/// combination rather than the two small ones here.
///
/// `present` gates this exactly like [`decide`] gates deploy-vs-repair on
/// the power axis: a `Paused` declaration on an absent agent reaches
/// `Converge::Deploy` above, and the pause marker converges on the *next*
/// pass once that deploy has made the agent present, not this one — writing
/// a pause marker into a harness dir that may not exist yet is the same
/// "repair against a guess" this module's `Offline`/`Destroyed` present-gate
/// already refuses on the power axis.
fn decide_pause(state: AgentState, present: bool, currently_paused: bool) -> PauseConverge {
    if !present {
        return PauseConverge::Nothing;
    }
    match state {
        AgentState::Paused if currently_paused => PauseConverge::Nothing,
        AgentState::Paused => PauseConverge::Pause,
        AgentState::Up if currently_paused => PauseConverge::Resume,
        // `Up` already unpaused: agreement. `Offline`/`Destroyed`: the pause
        // marker is moot either way (no turn loop runs on a stopped or gone
        // container to park), so it is left as-is rather than cleared —
        // whatever it says gets re-evaluated the moment either state moves
        // back to `Up` or `Paused`.
        AgentState::Up | AgentState::Offline | AgentState::Destroyed => PauseConverge::Nothing,
    }
}

#[cfg(test)]
mod tests {
    use std::collections::{BTreeMap, BTreeSet};

    use super::{Converge, PauseConverge, Plan, carries_a_declaration, decide, decide_pause, plan};
    use crate::power::Wanted;
    use swarm_wire::wanted::{AgentState, AgentWanted, HiveWanted};

    fn declaration(agents: &[(&str, AgentState)]) -> HiveWanted {
        HiveWanted {
            agents: agents
                .iter()
                .map(|(name, state)| ((*name).to_owned(), AgentWanted { state: *state }))
                .collect(),
        }
    }

    fn set(names: &[&str]) -> BTreeSet<String> {
        names.iter().map(|n| (*n).to_owned()).collect()
    }

    fn intents(entries: &[(&str, Option<Wanted>)]) -> BTreeMap<String, Option<Wanted>> {
        entries
            .iter()
            .map(|(name, intent)| ((*name).to_owned(), *intent))
            .collect()
    }

    fn paused(entries: &[(&str, bool)]) -> BTreeMap<String, bool> {
        entries
            .iter()
            .map(|(name, is_paused)| ((*name).to_owned(), *is_paused))
            .collect()
    }

    /// The property the whole migration rides on: this hive's own agents are
    /// not converged, they are not looked at. `adopted` is the control — the
    /// same call must still act on what the declaration *does* name, or this
    /// passes on a loop that plans nothing at all.
    #[test]
    fn an_agent_the_declaration_does_not_name_is_never_touched() {
        let declared = declaration(&[("adopted", AgentState::Up)]);
        let planned = plan(
            &declared,
            &set(&["adopted", "legacy"]),
            &intents(&[("adopted", Some(Wanted::Offline)), ("legacy", None)]),
            &paused(&[]),
        );
        assert_eq!(
            planned,
            Plan {
                deploy: vec![],
                start: vec!["adopted".to_owned()],
                stop: vec![],
                destroy: vec![],
                pause: vec![],
                resume: vec![],
            }
        );
    }

    /// An intent that could not be read is skipped rather than treated as "no
    /// row". `readable` is the control for the same call.
    #[test]
    fn an_unreadable_intent_skips_only_that_agent() {
        let declared = declaration(&[("unreadable", AgentState::Up), ("readable", AgentState::Up)]);
        let planned = plan(
            &declared,
            &set(&["unreadable", "readable"]),
            &intents(&[("readable", None)]),
            &paused(&[]),
        );
        assert_eq!(planned.start, vec!["readable".to_owned()]);
        assert!(planned.deploy.is_empty() && planned.stop.is_empty());
    }

    /// The pause axis does not gate on a readable intent row at all — a
    /// `paused` agent this hive has no power-intent row for still gets its
    /// pause marker converged, unlike the power axis right above.
    #[test]
    fn pause_converges_even_when_the_power_intent_is_unreadable() {
        let declared = declaration(&[("unreadable", AgentState::Paused)]);
        let planned = plan(
            &declared,
            &set(&["unreadable"]),
            &intents(&[]),
            &paused(&[("unreadable", false)]),
        );
        assert_eq!(planned.pause, vec!["unreadable".to_owned()]);
        assert!(planned.start.is_empty() && planned.deploy.is_empty());
    }

    /// A declared-paused agent that is present, running and already carries
    /// the marker needs nothing on either axis.
    #[test]
    fn an_already_paused_agent_converges_to_nothing() {
        let declared = declaration(&[("parked", AgentState::Paused)]);
        let planned = plan(
            &declared,
            &set(&["parked"]),
            &intents(&[("parked", Some(Wanted::Up))]),
            &paused(&[("parked", true)]),
        );
        assert_eq!(planned, Plan::default());
    }

    /// The case `decide_pause`'s own comment calls out: declared `Paused` but
    /// currently stopped needs a power `Start` *and* a `Pause`, in the same
    /// pass — this is the test that would fail if the two axes were folded
    /// into one enum instead of decided independently.
    #[test]
    fn a_stopped_agent_declared_paused_gets_both_a_start_and_a_pause() {
        let declared = declaration(&[("parked", AgentState::Paused)]);
        let planned = plan(
            &declared,
            &set(&["parked"]),
            &intents(&[("parked", Some(Wanted::Offline))]),
            &paused(&[("parked", false)]),
        );
        assert_eq!(planned.start, vec!["parked".to_owned()]);
        assert_eq!(planned.pause, vec!["parked".to_owned()]);
    }

    /// A declared-`Up` agent whose marker is still set from an earlier
    /// `Paused` declaration gets resumed.
    #[test]
    fn an_up_agent_still_carrying_the_marker_is_resumed() {
        let declared = declaration(&[("parked", AgentState::Up)]);
        let planned = plan(
            &declared,
            &set(&["parked"]),
            &intents(&[("parked", Some(Wanted::Up))]),
            &paused(&[("parked", true)]),
        );
        assert_eq!(planned.resume, vec!["parked".to_owned()]);
    }

    #[test]
    fn a_declared_agent_this_hive_does_not_have_is_deployed() {
        assert_eq!(decide(AgentState::Up, false, None), Converge::Deploy);
        // Even holding an intent: a row can outlive its container.
        assert_eq!(
            decide(AgentState::Up, false, Some(Wanted::Up)),
            Converge::Deploy
        );
    }

    #[test]
    fn a_hive_that_already_agrees_queues_nothing() {
        assert_eq!(
            decide(AgentState::Up, true, Some(Wanted::Up)),
            Converge::Nothing
        );
        assert_eq!(
            decide(AgentState::Offline, true, Some(Wanted::Offline)),
            Converge::Nothing
        );
    }

    #[test]
    fn a_disagreeing_intent_is_converged_both_ways() {
        assert_eq!(
            decide(AgentState::Up, true, Some(Wanted::Offline)),
            Converge::Start
        );
        assert_eq!(
            decide(AgentState::Offline, true, Some(Wanted::Up)),
            Converge::Stop
        );
    }

    /// No row is not agreement. An agent that predates the power store is
    /// converged to what it was declared as, not left on whatever it happens
    /// to be doing.
    #[test]
    fn an_agent_with_no_intent_row_is_converged() {
        assert_eq!(decide(AgentState::Up, true, None), Converge::Start);
        assert_eq!(decide(AgentState::Offline, true, None), Converge::Stop);
    }

    /// Declared offline, no container: the one absent case that must not
    /// deploy. Its control is the `Up` case above, which must.
    #[test]
    fn an_absent_agent_declared_offline_is_left_absent() {
        assert_eq!(decide(AgentState::Offline, false, None), Converge::Nothing);
    }

    /// A present agent declared destroyed gets torn down — the whole point of
    /// this state. Intent is irrelevant, unlike `Up`/`Offline`: even an
    /// agent whose power intent is `Up` still gets destroyed, since a
    /// terminal declaration outranks a steady-state one.
    #[test]
    fn a_present_agent_declared_destroyed_is_destroyed() {
        assert_eq!(decide(AgentState::Destroyed, true, None), Converge::Destroy);
        assert_eq!(
            decide(AgentState::Destroyed, true, Some(Wanted::Up)),
            Converge::Destroy
        );
    }

    /// Already gone: destroying an absent agent would be acting on nothing.
    /// Its control is the present case above, which must destroy.
    #[test]
    fn an_absent_agent_declared_destroyed_is_left_absent() {
        assert_eq!(
            decide(AgentState::Destroyed, false, None),
            Converge::Nothing
        );
    }

    /// `Paused` shares every `Up` arm on the power axis — same deploy-when-
    /// absent, same start-when-stopped behaviour.
    #[test]
    fn paused_converges_on_the_power_axis_exactly_like_up() {
        for present in [true, false] {
            for intent in [None, Some(Wanted::Up), Some(Wanted::Offline)] {
                assert_eq!(
                    decide(AgentState::Paused, present, intent),
                    decide(AgentState::Up, present, intent),
                    "present={present:?} intent={intent:?}"
                );
            }
        }
    }

    #[test]
    fn decide_pause_sets_the_marker_for_a_present_unpaused_paused_declaration() {
        assert_eq!(
            decide_pause(AgentState::Paused, true, false),
            PauseConverge::Pause
        );
    }

    #[test]
    fn decide_pause_is_idempotent_once_the_marker_already_matches() {
        assert_eq!(
            decide_pause(AgentState::Paused, true, true),
            PauseConverge::Nothing
        );
        assert_eq!(
            decide_pause(AgentState::Up, true, false),
            PauseConverge::Nothing
        );
    }

    #[test]
    fn decide_pause_clears_the_marker_for_a_declared_up_agent() {
        assert_eq!(
            decide_pause(AgentState::Up, true, true),
            PauseConverge::Resume
        );
    }

    /// No container, nothing to write a marker into — the same present-gate
    /// [`decide`] applies to `Offline`/`Destroyed`, mirrored here for the
    /// pause axis regardless of declared state.
    #[test]
    fn decide_pause_does_nothing_for_an_absent_agent() {
        for state in [
            AgentState::Up,
            AgentState::Paused,
            AgentState::Offline,
            AgentState::Destroyed,
        ] {
            for currently_paused in [true, false] {
                assert_eq!(
                    decide_pause(state, false, currently_paused),
                    PauseConverge::Nothing,
                    "state={state:?} currently_paused={currently_paused:?}"
                );
            }
        }
    }

    /// `Offline`/`Destroyed` never touch the marker either way — it is left
    /// exactly as found, whatever that is.
    #[test]
    fn decide_pause_leaves_offline_and_destroyed_alone() {
        for state in [AgentState::Offline, AgentState::Destroyed] {
            for currently_paused in [true, false] {
                assert_eq!(
                    decide_pause(state, true, currently_paused),
                    PauseConverge::Nothing,
                    "state={state:?} currently_paused={currently_paused:?}"
                );
            }
        }
    }

    /// The watch's half of "absence is not a deletion order". `Put` is the
    /// control: without it this would pass on a function that refused
    /// everything, which would silently stop the fast path converging at all.
    #[test]
    fn only_a_put_carries_a_declaration_to_converge_to() {
        use async_nats::jetstream::kv::Operation;

        assert!(carries_a_declaration(Operation::Put));
        for withdrawn in [Operation::Delete, Operation::Purge] {
            assert!(
                !carries_a_declaration(withdrawn),
                "{withdrawn:?} must not converge"
            );
        }
    }

    /// Version skew is handled one layer up, at the decode: `AgentState` is
    /// closed, so an unknown value never reaches [`decide`] — it fails the
    /// whole declaration in `swarm-queue-client`, which owns that test
    /// (`an_unknown_state_fails_the_whole_declaration`). There is deliberately
    /// no case for it here: a test asserting something unrepresentable would
    /// pass forever without measuring anything.
    #[test]
    fn every_state_this_build_knows_is_covered_above() {
        for state in [
            AgentState::Up,
            AgentState::Offline,
            AgentState::Paused,
            AgentState::Destroyed,
        ] {
            let seen = [true, false].iter().any(|present| {
                decide(state, *present, None) != Converge::Nothing
                    || decide(state, *present, Some(Wanted::Up)) != Converge::Nothing
            });
            assert!(
                seen,
                "{state:?} produces no power action in any combination"
            );
        }
    }

    /// The pause-axis equivalent of the test above: every state this build
    /// knows produces a non-`Nothing` pause decision in at least one
    /// combination, so a state that silently fell out of `decide_pause`
    /// would be caught here rather than passing forever unmeasured. `Up` and
    /// `Paused` are the two expected to; `Offline`/`Destroyed` are asserted
    /// separately above as deliberately inert on this axis.
    #[test]
    fn up_and_paused_both_produce_a_pause_action_in_some_combination() {
        for state in [AgentState::Up, AgentState::Paused] {
            let seen = [true, false].iter().any(|currently_paused| {
                decide_pause(state, true, *currently_paused) != PauseConverge::Nothing
            });
            assert!(
                seen,
                "{state:?} produces no pause action in any combination"
            );
        }
    }

    /// A client that never connects: initial-connect retries run in the
    /// background and `ensure_connected` keeps failing.
    async fn never_connected() -> async_nats::Client {
        async_nats::ConnectOptions::new()
            .retry_on_initial_connect()
            .connect("nats://127.0.0.1:1")
            .await
            .expect("a client that retries its first connect is returned unconnected")
    }

    /// One deadline bounds the whole connect wait: time already spent before
    /// `wait_connected` counts against it, so `pull`'s client wait and its
    /// connection wait together stay inside one `CONNECT_WAIT`.
    #[tokio::test(start_paused = true)]
    async fn the_connect_wait_is_bounded_by_one_deadline() {
        let client = never_connected().await;
        let start = tokio::time::Instant::now();
        let deadline = start + super::CONNECT_WAIT;
        tokio::time::sleep(std::time::Duration::from_secs(59)).await;
        assert!(super::wait_connected(&client, deadline).await.is_err());
        assert!(
            start.elapsed() <= super::CONNECT_WAIT + super::CONNECT_POLL,
            "waited {:?} in total",
            start.elapsed()
        );
    }

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

        use super::super::{MAX_RETRY_DELAY, SessionEnd, attach_then_pull, retry_delay, supervise};

        #[tokio::test]
        async fn the_watch_attaches_before_the_pull_runs() {
            let log = Arc::new(Mutex::new(Vec::new()));
            let (a, p) = (Arc::clone(&log), Arc::clone(&log));
            let got = attach_then_pull(
                async move {
                    a.lock().unwrap().push("attach");
                    Some(7)
                },
                async move {
                    p.lock().unwrap().push("pull");
                    Ok(())
                },
            )
            .await;
            assert_eq!(got, Some(7));
            assert_eq!(*log.lock().unwrap(), ["attach", "pull"]);
        }

        #[tokio::test]
        async fn a_failed_attach_does_not_pull() {
            let log = Arc::new(Mutex::new(Vec::new()));
            let p = Arc::clone(&log);
            let got: Option<u8> = attach_then_pull(async { None }, async move {
                p.lock().unwrap().push("pull");
                Ok(())
            })
            .await;
            assert_eq!(got, None);
            assert!(log.lock().unwrap().is_empty(), "no watch, no pull");
        }

        #[tokio::test]
        async fn a_failed_pull_keeps_the_watch() {
            let got = attach_then_pull(async { Some(1) }, async { anyhow::bail!("down") }).await;
            assert_eq!(got, Some(1));
        }

        #[tokio::test(start_paused = true)]
        async fn every_re_attach_pulls_again() {
            let log = Arc::new(Mutex::new(Vec::new()));
            let (_tx, shutdown) = tokio::sync::watch::channel(false);
            let l = Arc::clone(&log);
            supervise(
                move || {
                    let l = Arc::clone(&l);
                    async move {
                        let (a, p) = (Arc::clone(&l), Arc::clone(&l));
                        let _ = attach_then_pull(
                            async move {
                                a.lock().unwrap().push("attach");
                                Some(())
                            },
                            async move {
                                p.lock().unwrap().push("pull");
                                Ok(())
                            },
                        )
                        .await;
                        if l.lock().unwrap().len() >= 6 {
                            SessionEnd::Shutdown
                        } else {
                            SessionEnd::Lost { attached: true }
                        }
                    }
                },
                shutdown,
            )
            .await;
            assert_eq!(
                *log.lock().unwrap(),
                ["attach", "pull", "attach", "pull", "attach", "pull"]
            );
        }

        #[tokio::test(start_paused = true)]
        async fn failed_attaches_are_retried_not_abandoned() {
            let tries = Arc::new(Mutex::new(0_u32));
            let (_tx, shutdown) = tokio::sync::watch::channel(false);
            let t = Arc::clone(&tries);
            supervise(
                move || {
                    let t = Arc::clone(&t);
                    async move {
                        let mut n = t.lock().unwrap();
                        *n += 1;
                        if *n >= 5 {
                            SessionEnd::Shutdown
                        } else {
                            SessionEnd::Lost { attached: false }
                        }
                    }
                },
                shutdown,
            )
            .await;
            assert_eq!(*tries.lock().unwrap(), 5);
        }

        #[tokio::test(start_paused = true)]
        async fn an_attach_resets_the_backoff() {
            let start = tokio::time::Instant::now();
            let starts = Arc::new(Mutex::new(Vec::new()));
            let (_tx, shutdown) = tokio::sync::watch::channel(false);
            let s = Arc::clone(&starts);
            supervise(
                move || {
                    let s = Arc::clone(&s);
                    async move {
                        let mut s = s.lock().unwrap();
                        s.push(start.elapsed().as_secs());
                        match s.len() {
                            1 | 2 => SessionEnd::Lost { attached: false },
                            3 => SessionEnd::Lost { attached: true },
                            _ => SessionEnd::Shutdown,
                        }
                    }
                },
                shutdown,
            )
            .await;
            // Waits: 10 and 20 after two misses, then 5 once an attach worked.
            assert_eq!(*starts.lock().unwrap(), [0, 10, 30, 35]);
        }

        #[test]
        fn the_retry_delay_doubles_and_is_capped() {
            let secs: Vec<u64> = (0..8).map(|n| retry_delay(n).as_secs()).collect();
            assert_eq!(secs, [5, 10, 20, 40, 80, 160, 300, 300]);
            assert_eq!(retry_delay(u32::MAX), MAX_RETRY_DELAY);
            assert!(retry_delay(1_000) <= Duration::from_mins(5));
        }
    }
}