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_queue_client::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, because the controller owns that.

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

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

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

/// 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".
///
/// # 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 Some(client) = crate::swarm_queue::client().await else {
        // Absent or failed — either way already reported by the shared
        // connector (an `info` log or a `swarm_queue_config` banner).
        return Ok(());
    };
    // An unconnected client does not fail a JetStream request, it hangs on it.
    // A hung node is worse than a failed one: the dashboard shows it running.
    swarm_queue_client::ensure_connected(&client)?;

    let Some(store) = swarm_queue_client::wanted::open_read_only(&client, &hive).await 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
}

/// Converge again each time the controller republishes this hive's declaration.
///
/// The **fast** path, and it does not replace [`pull`]: this hears only what is
/// published while it is listening, so a hive that was down still learns the
/// current declaration from the boot read. Both, not either.
///
/// ⚠️ 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 `None` 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,
    mut shutdown: tokio::sync::watch::Receiver<bool>,
) {
    use futures_util::StreamExt as _;

    let Some(mut updates) = swarm_queue_client::wanted::watch(&client, &hive).await else {
        tracing::warn!(
            %hive,
            "wanted state: cannot watch this hive's bucket; changes will be \
             picked up at the next boot instead"
        );
        return;
    };
    tracing::info!(%hive, "wanted state: watching for declarations");

    loop {
        tokio::select! {
            entry = updates.next() => {
                let Some(entry) = entry else {
                    // Same reading as the sibling subscriptions: `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");
                    return;
                };
                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;
            }
        }
    }
}

/// 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(_) => queued = true,
            // 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_queue_client::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"
            );
        }
    }
}