kanade_shared/

nats_client.rs

//! Shared NATS client constructor.
//!
//! Every binary names the [`NatsRole`] it connects as, and the token is
//! resolved per role (first match wins):
//!
//!   1. Windows registry — `HKLM\SOFTWARE\kanade\<role>\NatsToken`
//!      (`REG_SZ`). The role-specific credential. Hardened ACL (SYSTEM +
//!      Admin only) keeps the token out of low-privilege users' reach,
//!      which Machine-scope env vars cannot do.
//!   2. Windows registry — `HKLM\SOFTWARE\kanade\agent\NatsToken`. The
//!      **shared** credential every role used before roles existed. Kept as
//!      a fallback so an existing deployment keeps working untouched; see
//!      "Staged migration" below.
//!   3. `$KANADE_NATS_TOKEN` environment variable. Dev / fallback path. The
//!      agent service runs as LocalSystem so user-session env vars never
//!      reach it; this branch only fires for `cargo run` / interactive
//!      shells.
//!   4. No token — connect unauthenticated. Works against a broker started
//!      without `authorization { … }`.
//!
//! # Per-role users, with the token as fallback
//!
//! A role may also hold a NATS *user*: `NatsUser` and `NatsPassword`
//! (`REG_SZ`) under its **own** registry key, or `$KANADE_NATS_USER` /
//! `$KANADE_NATS_PASSWORD`. There is deliberately no shared user — the shared
//! credential is the token. A pair with only one half is a configuration
//! error that fails the connect, never a silent fallback.
//!
//! With no user provisioned nothing below applies: the connection presents
//! the token (or nothing) exactly as before and no probe is ever made.
//!
//! With a user provisioned the broker may be on either side of the
//! token → `users` switch, and the switch is atomic on the broker, so the
//! credential is chosen per connection attempt (initial and every reconnect)
//! from an auth callback. Whether a rejected attempt is retried is up to the
//! async-nats version (0.50 keeps retrying, and re-runs the callback, while
//! another version may end the connection task for good, leaving a `Client`
//! that never talks again), so "present the user, and on rejection retry with
//! the token" is not left to its retry loop. Instead each attempt first opens
//! a short-lived *probe* connection with the user, which does not retry:
//!
//!   * accepted → present the user;
//!   * rejected (authorization violation) → present the token, or fail
//!     naming the missing token if there is none;
//!   * anything else (unreachable, timeout) → defer the attempt with an
//!     authentication callback error. A cached answer describes an old broker
//!     mode and must not become a refused handshake after a restart.
//!
//! The process keeps one `Client`: a deferred attempt is retried and re-probed.
//! That a refusal is retried and never ends the connection task is pinned
//! against a real broker for the pinned async-nats, so a version change that
//! breaks it fails a test instead of leaving a silent client.
//!
//! What a switch or restart guarantees is stated as [`RESUME_BOUND`] and
//! [`EXIT_BOUND`].
//! A client whose connection task has terminated for any reason (a panic in
//! it, a drain, a version that treats a violation as terminal) is detected by
//! [`wait_until_dead`] so its process can exit for a supervised restart. The
//! same call also reports a broker that refuses the credential on every
//! attempt for a sustained period, since async-nats would otherwise retry it
//! forever and leave a process that looks alive but can never talk.
//!
//! # Why roles exist here (#1155)
//!
//! The broker authorises a *connection*, and a connection is only as
//! specific as the credential that opened it. While every binary presented
//! the same token, the broker could not tell an agent from the backend from
//! the CLI, so no `permissions` block could say "only the backend may
//! subscribe `remote.frame.>`" — there was nothing to hang the rule on.
//! That is why a shared token means a token holder can execute code on any
//! endpoint and silently watch any remote-assistance session (#1140).
//!
//! Distinct credentials do not fix that by themselves; the broker config has
//! to grow the matching `authorization { users: [...] }` entries. This
//! module is the half that makes those entries *expressible*.
//!
//! # Staged migration
//!
//! Step 2 above is the whole migration strategy. A fleet running today has
//! one token, provisioned at `…\kanade\agent\NatsToken` on every host
//! regardless of role. After this change it keeps working: no role key
//! exists, so every role falls through to the shared one and presents
//! exactly what it presented before.
//!
//! Rolling out per-role credentials is then per-host and reversible — write
//! `…\kanade\backend\NatsToken` on the backend host and it starts using it;
//! delete it and it falls back. The broker only needs to start
//! *distinguishing* the roles once every host has its own, so the config
//! change lands last, when it can no longer lock anyone out.
//!
//! No deploy script writes a role key yet: `deploy-backend.ps1` still
//! provisions the shared path, so today the role key is a manual registry
//! write. That is deliberate — the scripted path should start writing role
//! keys in the same change that teaches the broker to tell the roles apart,
//! because until then a role key changes nothing and a script that writes
//! only the role key (dropping the shared one) would strand the CLI on a
//! backend-only host.
//!
//! The order matters and is deliberate: role key first, shared key second.
//! The reverse would make the shared token permanent — a host that still has
//! it (all of them, today) would never notice its role key.
//!
//! # What the broker will and will not accept (measured, #1270)
//!
//! Two nats-server behaviours constrain every plan built on this module, so
//! they are recorded here rather than rediscovered:
//!
//! * A config may not carry **both** a `token` and a `users` array —
//!   nats-server refuses to start: *"Can not have a token and a users
//!   array"*. And once `users` are defined, a client presenting a token is
//!   rejected with an Authorization Violation, even when the token equals a
//!   user's password. So the shared token and a per-role `users` split
//!   cannot coexist for a transition window: the flip is atomic, and every
//!   host must already hold a credential of the new shape before it happens.
//!   Resolving a *token* per role, which is all this module does today, is
//!   therefore not sufficient for that split — the client has to learn to
//!   present a user as well.
//! * `/connz?auth=1` reports `authorized_user` per connection. Under
//!   `users` that is the username — the per-host answer #1270 wants. Under
//!   `token`, nats-server 2.14.3 reports the literal `[REDACTED]`: it hides
//!   the credential, so token mode can say *that* a host is on the shared
//!   token but never anything finer. Whether the value is hidden is the
//!   broker build's choice, not ours, so a consumer must assume it may be
//!   handling a secret; see [`CredentialProbe`] for the one question it can
//!   safely ask about one.
//!
//! # Limits worth naming
//!
//! A per-role token still cannot express per-*agent* identity. A role
//! credential permitted to subscribe `commands.pc.*` lets any agent holding
//! it read another agent's inbox. This narrows a fleet-wide compromise to a
//! fleet-wide **agent-role** compromise, which is better, not solved. The
//! end state is per-agent identity (NKeys / NATS-JWT), for which the plan is
//! to grow `ConnectOptions` here so every binary picks up the upgrade for
//! free. Same for mTLS.

use std::collections::HashMap;
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex, OnceLock, Weak};
use std::time::Duration;

use anyhow::{Context, Result, bail};
use tracing::{debug, info, warn};

use crate::secrets;

const ENV_TOKEN: &str = "KANADE_NATS_TOKEN";
const REG_VALUE: &str = "NatsToken";
const ENV_USER: &str = "KANADE_NATS_USER";
const ENV_PASSWORD: &str = "KANADE_NATS_PASSWORD";
const REG_USER: &str = "NatsUser";
const REG_PASSWORD: &str = "NatsPassword";

/// How long the credential probe may take. Short on purpose: it runs inside
/// the reconnect path, so a black-holed broker must not stall every attempt.
const PROBE_TIMEOUT: Duration = Duration::from_secs(3);

/// Name of the probe connection. Deliberately *not* a `kanade-` name: the
/// backend attributes `kanade-<role>` connections to hosts, and a transient
/// probe must not show up there as an unknown role.
const PROBE_NAME: &str = "auth-probe";

/// How often the client sends a protocol PING when idle. Set explicitly (it
/// equals async-nats' default) because every liveness bound below is derived
/// from it: a healthy idle connection receives a PONG at least this often.
const PING_INTERVAL: Duration = Duration::from_secs(60);

/// How often [`wait_until_dead`] runs its local check. The check only reads
/// counters and queues a flush on the local command channel, so a short
/// interval costs no network traffic.
const CHECK_INTERVAL: Duration = Duration::from_secs(5);

/// How long without any received byte before the client is suspected stalled:
/// three ping periods, so a healthy idle connection (one PONG per
/// [`PING_INTERVAL`]) is two full periods away from being suspected.
const STALL_BOUND: Duration = Duration::from_secs(3 * PING_INTERVAL.as_secs());

const HEALTH_TIMEOUT: Duration = Duration::from_secs(5);

/// The longest async-nats waits between reconnect attempts (its backoff cap).
const RECONNECT_DELAY_MAX_SECS: u64 = 4;

/// async-nats' default timeout for one connection handshake.
const CONNECT_TIMEOUT_SECS: u64 = 5;

/// The longest a single reconnect attempt can take: the backoff, then the
/// credential probe (its own timeout plus the guard around it), then the
/// handshake the probe's answer decides.
const ATTEMPT_MAX_SECS: u64 =
    RECONNECT_DELAY_MAX_SECS + PROBE_TIMEOUT.as_secs() + 1 + CONNECT_TIMEOUT_SECS;

/// How long after the broker starts answering in its new authentication mode
/// (a restart, or a switch between `token` and `users`) a client may take to
/// talk again: two attempts, allowing a probe interrupted by the switch,
/// plus slack. Both bounds are
/// counted from the moment the broker answers, not from the moment it went
/// away; a broker that is down is waited for indefinitely.
pub const RESUME_BOUND: Duration = Duration::from_secs(2 * ATTEMPT_MAX_SECS + 4);

/// How long after the broker starts answering in its new mode a client that
/// has *not* resumed may stay alive before its process exits non-zero for the
/// service manager to restart: the stall threshold, then the next local
/// check, a bounded flush, and a bounded credential witness. The service
/// manager's own restart delay comes on top. A broker outage seen by the
/// witness only moves the threshold to [`RESUME_BOUND`] after it ends, which
/// is never later than this.
pub const EXIT_BOUND: Duration = Liveness::PRODUCTION.exit_bound();

const _: () = {
    assert!(STALL_BOUND.as_secs() >= PING_INTERVAL.as_secs() + 60);
    assert!(RESUME_BOUND.as_secs() <= STALL_BOUND.as_secs());
    assert!(EXIT_BOUND.as_secs() < 300);
};

/// The timing of [`wait_until_dead_with`]. Only [`Liveness::PRODUCTION`] is
/// meant for shipped code; other values exist so tests can shorten the
/// bounds, and a `stall_bound` at or below the ping interval would judge a
/// healthy idle connection stalled.
#[derive(Debug, Clone, Copy)]
pub struct Liveness {
    /// How often the local check runs.
    pub check_interval: Duration,
    /// How long without received bytes before a credential witness is tried.
    pub stall_bound: Duration,
}

impl Liveness {
    pub const PRODUCTION: Liveness = Liveness {
        check_interval: CHECK_INTERVAL,
        stall_bound: STALL_BOUND,
    };

    /// The longest a stalled client stays alive after the broker answers.
    pub const fn exit_bound(&self) -> Duration {
        Duration::from_secs(
            self.stall_bound.as_secs()
                + self.check_interval.as_secs()
                + 2 * HEALTH_TIMEOUT.as_secs(),
        )
    }

    /// The grace given after a witness finds the broker reachable again.
    fn resume_grace(&self) -> Duration {
        RESUME_BOUND.min(self.stall_bound)
    }
}

/// What the supervisor should do after a credential witness.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Verdict {
    /// Keep watching.
    Wait,
    /// The broker answered and the client is still silent: exit, unless a
    /// recheck shows progress.
    Stalled,
}

/// The stall decision as a pure state machine over caller-supplied times, so
/// the long-window behaviour is testable without waiting.
struct StallJudge {
    liveness: Liveness,
    deadline: tokio::time::Instant,
    /// The last witness found the broker unreachable. The silence then
    /// describes the outage, not the client.
    outage: bool,
    next_witness: Option<tokio::time::Instant>,
}

impl StallJudge {
    fn new(liveness: Liveness, now: tokio::time::Instant) -> Self {
        Self {
            liveness,
            deadline: now + liveness.stall_bound,
            outage: false,
            next_witness: None,
        }
    }

    /// Receive progress was observed: the clock restarts and any outage
    /// verdict no longer applies.
    fn progressed(&mut self, now: tokio::time::Instant) {
        self.deadline = now + self.liveness.stall_bound;
        self.outage = false;
        self.next_witness = None;
    }

    /// Whether a witness connection is due. Never true before the deadline.
    fn witness_due(&self, now: tokio::time::Instant) -> bool {
        now >= self.deadline && self.next_witness.is_none_or(|at| now >= at)
    }

    fn witnessed(&mut self, now: tokio::time::Instant, outcome: ProbeOutcome) -> Verdict {
        if outcome == ProbeOutcome::Unreachable {
            self.outage = true;
            self.next_witness = Some(now + self.liveness.resume_grace());
            return Verdict::Wait;
        }
        if std::mem::take(&mut self.outage) {
            // First sight of the broker after an outage: the client may still
            // be in its reconnect backoff, so give it the resume grace.
            self.deadline = self.deadline.max(now + self.liveness.resume_grace());
            self.next_witness = None;
            return Verdict::Wait;
        }
        Verdict::Stalled
    }
}

/// A connection that has been refused this many times in a row, over at
/// least [`AUTH_REJECTION_WINDOW`], with no successful connect in between, is
/// treated as failed. async-nats retries a refused credential forever, which
/// leaves a process that can never talk but looks alive; the window keeps a
/// broker that is merely mid-reload (a few quick refusals) from counting.
const AUTH_REJECTION_LIMIT: usize = 5;
const AUTH_REJECTION_WINDOW: Duration = Duration::from_secs(10);

/// Prefix every kanade connection announces in its `name`.
const NAME_PREFIX: &str = "kanade-";

/// Separator between the role and the host identity in a connection name.
/// `/` is safe as a delimiter because the identity is a Windows computer
/// name, which cannot contain one.
const NAME_SEP: char = '/';

/// Registry subkey holding the pre-#1155 shared credential. Also the agent's
/// role key, which is not a coincidence — the shared token was provisioned
/// under the agent's path because agents were the first thing to need it.
const REG_SHARED_SUBKEY: &str = r"SOFTWARE\kanade\agent";

/// Which kanade binary is opening the connection.
///
/// Named on every call rather than inferred, because the broker's view of a
/// connection comes entirely from the credential it presents: a caller that
/// picks the wrong role does not get a warning, it gets someone else's
/// permissions.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NatsRole {
    /// The endpoint agent. The most numerous and least trusted role — one
    /// compromised endpoint holds this credential.
    Agent,
    /// The backend. The only role that needs to see the whole fleet.
    Backend,
    /// The operator CLI, including the backend-down recovery path that
    /// drives agents over NATS directly.
    Cli,
}

impl NatsRole {
    pub fn as_str(self) -> &'static str {
        match self {
            NatsRole::Agent => "agent",
            NatsRole::Backend => "backend",
            NatsRole::Cli => "cli",
        }
    }

    /// Registry subkey holding this role's credential.
    fn reg_subkey(self) -> String {
        format!(r"SOFTWARE\kanade\{}", self.as_str())
    }
}

/// Resolve a role's token, given a registry reader and the environment
/// fallback.
///
/// Split from [`resolve_token`] so the *ordering* — the part that decides
/// whether a migration is reversible — is testable without a Windows
/// registry to write to.
fn resolve_token_with(
    role: NatsRole,
    read_reg: impl Fn(&str, &str) -> Option<String>,
    env: Option<String>,
) -> Option<String> {
    if let Some(t) = read_reg(&role.reg_subkey(), REG_VALUE) {
        return Some(t);
    }
    if let Some(t) = read_reg(REG_SHARED_SUBKEY, REG_VALUE) {
        return Some(t);
    }
    env.filter(|t| !t.is_empty())
}

fn resolve_token(role: NatsRole) -> Option<String> {
    resolve_token_with(
        role,
        secrets::read_hklm_value,
        std::env::var(ENV_TOKEN).ok(),
    )
}

/// A named NATS user. The password is never shown by `Debug`.
#[derive(Clone, PartialEq, Eq)]
struct UserCredential {
    name: String,
    password: String,
}

impl std::fmt::Debug for UserCredential {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("UserCredential")
            .field("name", &self.name)
            .field("password", &"<redacted>")
            .finish()
    }
}

/// Turn the two halves of a user pair into a credential, or a configuration
/// error naming what is missing and where (never a value).
fn pair(
    name: Option<String>,
    password: Option<String>,
    source: &str,
) -> Result<Option<UserCredential>> {
    match (name, password) {
        (Some(name), Some(password)) => Ok(Some(UserCredential { name, password })),
        (None, None) => Ok(None),
        (Some(_), None) => bail!("NATS user is set but its password is missing ({source})"),
        (None, Some(_)) => bail!("NATS password is set but its user name is missing ({source})"),
    }
}

/// Resolve a role's user pair: its own registry key first, then the
/// environment. Unlike the token there is no shared-key step.
///
/// The registry is judged as a whole before the environment is consulted: a
/// half pair there is an error even when the environment is complete,
/// because quietly mixing sources would hide a half-provisioned host.
fn resolve_user_with(
    role: NatsRole,
    read_reg: impl Fn(&str, &str) -> Option<String>,
    env_user: Option<String>,
    env_password: Option<String>,
) -> Result<Option<UserCredential>> {
    let subkey = role.reg_subkey();
    let reg_user = read_reg(&subkey, REG_USER);
    let reg_password = read_reg(&subkey, REG_PASSWORD);
    if reg_user.is_some() || reg_password.is_some() {
        let source = format!(r"HKLM\{subkey}: {REG_USER} / {REG_PASSWORD}");
        return pair(reg_user, reg_password, &source);
    }
    let source = format!("${ENV_USER} / ${ENV_PASSWORD}");
    pair(
        env_user.filter(|v| !v.is_empty()),
        env_password.filter(|v| !v.is_empty()),
        &source,
    )
}

fn resolve_user(role: NatsRole) -> Result<Option<UserCredential>> {
    resolve_user_with(
        role,
        secrets::read_hklm_value,
        std::env::var(ENV_USER).ok(),
        std::env::var(ENV_PASSWORD).ok(),
    )
}

/// The credentials a connection is built from.
///
/// [`connect`] resolves them from the registry / environment; this type
/// exists so a caller (tests, chiefly) can supply them directly without
/// touching process-global state. `Debug` never prints a secret.
#[derive(Clone, Default)]
pub struct NatsCredentials {
    token: Option<String>,
    user: Option<UserCredential>,
}

impl NatsCredentials {
    pub fn new(token: Option<String>, user: Option<(String, String)>) -> Self {
        Self {
            token,
            user: user.map(|(name, password)| UserCredential { name, password }),
        }
    }

    fn resolve(role: NatsRole) -> Result<Self> {
        Ok(Self {
            token: resolve_token(role),
            user: resolve_user(role)?,
        })
    }
}

impl std::fmt::Debug for NatsCredentials {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("NatsCredentials")
            .field("token", &self.token.as_ref().map(|_| "<redacted>"))
            .field("user", &self.user)
            .finish()
    }
}

/// Which credential a connection attempt presents.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Choice {
    User,
    Token,
}

impl Choice {
    fn label(self) -> &'static str {
        match self {
            Choice::User => "user",
            Choice::Token => "token",
        }
    }
}

/// What the probe connection learned about the broker.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ProbeOutcome {
    Accepted,
    Rejected,
    Unreachable,
}

impl ProbeOutcome {
    fn label(self) -> &'static str {
        match self {
            ProbeOutcome::Accepted => "accepted",
            ProbeOutcome::Rejected => "rejected",
            ProbeOutcome::Unreachable => "unreachable",
        }
    }
}

/// The credential the broker last *proved* it accepts or rejects for a role,
/// shared between the connection (which writes it on every attempt) and the
/// [`CredentialProbe`] (which reads it). Only a probe answer is recorded — a
/// guess made while the broker was unreachable is not evidence.
#[derive(Default)]
struct Live {
    decision: Mutex<Option<Choice>>,
    /// Consecutive authentication refusals: when the first one happened and
    /// how many there have been. Cleared by a successful connect.
    refusals: Mutex<Option<(std::time::Instant, usize)>>,
}

impl Live {
    /// Track the connection's events for [`Live::auth_failed`].
    fn observe(&self, ev: &async_nats::Event) {
        let mut refusals = self.refusals.lock().unwrap_or_else(|e| e.into_inner());
        match ev {
            // A successful connect, a drop, or any other kind of failed
            // attempt (broker unreachable, timeout) ends a run of refusals:
            // only an unbroken run is evidence the credential itself is bad,
            // and an offline broker is something to wait out.
            async_nats::Event::Connected | async_nats::Event::Disconnected => *refusals = None,
            async_nats::Event::ClientError(async_nats::ClientError::Other(kind))
                if *kind == async_nats::ConnectErrorKind::AuthorizationViolation.to_string() =>
            {
                let (first, n) = refusals.unwrap_or((std::time::Instant::now(), 0));
                *refusals = Some((first, n + 1));
            }
            // Callback errors include inconclusive probes, not broker refusals.
            async_nats::Event::ClientError(async_nats::ClientError::Other(kind))
                if *kind == async_nats::ConnectErrorKind::Authentication.to_string() => {}
            async_nats::Event::ClientError(_) => *refusals = None,
            _ => {}
        }
    }

    /// Whether the broker has refused every attempt for long enough that
    /// waiting longer is not going to help.
    fn auth_failed(&self) -> bool {
        match *self.refusals.lock().unwrap_or_else(|e| e.into_inner()) {
            Some((first, n)) => {
                n >= AUTH_REJECTION_LIMIT && first.elapsed() >= AUTH_REJECTION_WINDOW
            }
            None => false,
        }
    }

    fn get(&self) -> Option<Choice> {
        *self.decision.lock().unwrap_or_else(|e| e.into_inner())
    }

    fn set(&self, c: Choice) {
        *self.decision.lock().unwrap_or_else(|e| e.into_inner()) = Some(c);
    }
}

/// Process-wide per-role state, so [`CredentialProbe::for_role`] sees the
/// decision the already-open connection made without the callers having to
/// thread anything through.
fn role_lives() -> &'static Mutex<HashMap<&'static str, Arc<Live>>> {
    static LIVE: OnceLock<Mutex<HashMap<&'static str, Arc<Live>>>> = OnceLock::new();
    LIVE.get_or_init(Default::default)
}

fn publish_role_live(role: NatsRole, live: &Live) {
    // Preserve the reporting handle held by CredentialProbe::for_role.
    let reported = live_for(role);
    *reported.decision.lock().unwrap_or_else(|e| e.into_inner()) = live.get();
}

fn live_for(role: NatsRole) -> Arc<Live> {
    role_lives()
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .entry(role.as_str())
        .or_default()
        .clone()
}

struct ConnectionHealth {
    live: Arc<Live>,
    url: String,
    creds: NatsCredentials,
}

struct HealthEntry {
    stats: Weak<async_nats::client::Statistics>,
    health: Arc<ConnectionHealth>,
}

fn health_entries() -> &'static Mutex<Vec<HealthEntry>> {
    static HEALTH: OnceLock<Mutex<Vec<HealthEntry>>> = OnceLock::new();
    HEALTH.get_or_init(Default::default)
}

fn register_health(client: &async_nats::Client, health: Arc<ConnectionHealth>) {
    let mut entries = health_entries().lock().unwrap_or_else(|e| e.into_inner());
    entries.retain(|entry| entry.stats.strong_count() > 0);
    entries.push(HealthEntry {
        stats: Arc::downgrade(&client.statistics()),
        health,
    });
}

fn connection_health(client: &async_nats::Client) -> Option<Arc<ConnectionHealth>> {
    let stats = client.statistics();
    health_entries()
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .iter()
        .find(|entry| entry.stats.ptr_eq(&Arc::downgrade(&stats)))
        .map(|entry| entry.health.clone())
}

/// A fresh handshake proves the broker can accept this client's credentials.
/// It never retries, and neither its traffic nor its errors advance the
/// original client's progress clock or refusal accounting.
async fn witness(health: &ConnectionHealth) -> ProbeOutcome {
    let attempt = async {
        let mut opts = async_nats::ConnectOptions::new()
            .name("auth-witness")
            .connection_timeout(HEALTH_TIMEOUT);
        match &health.creds.user {
            Some(user) => match probe_user(&health.url, user).await {
                ProbeOutcome::Accepted => {
                    opts = opts.user_and_password(user.name.clone(), user.password.clone());
                }
                ProbeOutcome::Rejected => match &health.creds.token {
                    Some(token) => opts = opts.token(token.clone()),
                    None => return ProbeOutcome::Rejected,
                },
                ProbeOutcome::Unreachable => return ProbeOutcome::Unreachable,
            },
            None => {
                if let Some(token) = &health.creds.token {
                    opts = opts.token(token.clone());
                }
            }
        }
        match opts.connect(&health.url).await {
            Ok(_) => ProbeOutcome::Accepted,
            Err(error) if error.kind() == async_nats::ConnectErrorKind::AuthorizationViolation => {
                ProbeOutcome::Rejected
            }
            Err(_) => ProbeOutcome::Unreachable,
        }
    };
    tokio::time::timeout(HEALTH_TIMEOUT, attempt)
        .await
        .unwrap_or(ProbeOutcome::Unreachable)
}

/// Pick the credential for one attempt from the probe's answer.
///
/// Pure so the whole decision table is testable without a broker. The cached
/// decision is intentionally ignored: it describes a previous broker mode.
fn select(
    outcome: ProbeOutcome,
    _last_worked: Option<Choice>,
    have_token: bool,
) -> std::result::Result<Choice, &'static str> {
    match outcome {
        ProbeOutcome::Accepted => Ok(Choice::User),
        ProbeOutcome::Rejected if have_token => Ok(Choice::Token),
        ProbeOutcome::Rejected => {
            Err("the broker rejected the NATS user and no NatsToken is provisioned")
        }
        ProbeOutcome::Unreachable => {
            Err("NATS credential probe inconclusive; deferring this attempt")
        }
    }
}

/// Try the user on a connection of its own, which does not retry, so a
/// rejection costs this throwaway connection and nothing else.
///
/// Runs on a spawned task because the auth callback's future must be `Sync`
/// and async-nats' connect future is not; a join handle is.
async fn probe_user(url: &str, user: &UserCredential) -> ProbeOutcome {
    let (url, name, password) = (url.to_string(), user.name.clone(), user.password.clone());
    let probe = tokio::spawn(async move {
        let attempt = async_nats::ConnectOptions::new()
            .name(PROBE_NAME)
            .connection_timeout(PROBE_TIMEOUT)
            .user_and_password(name, password)
            .connect(url);
        match tokio::time::timeout(PROBE_TIMEOUT + Duration::from_secs(1), attempt).await {
            Ok(Ok(_client)) => ProbeOutcome::Accepted,
            Ok(Err(e)) if e.kind() == async_nats::ConnectErrorKind::AuthorizationViolation => {
                ProbeOutcome::Rejected
            }
            Ok(Err(_)) | Err(_) => ProbeOutcome::Unreachable,
        }
    });
    probe.await.unwrap_or(ProbeOutcome::Unreachable)
}

/// Probe, decide, record. `probe` is injected so the unit tests can drive
/// every outcome.
async fn choose<P, Fut>(
    role: NatsRole,
    have_token: bool,
    live: &Live,
    probe: P,
) -> std::result::Result<Choice, &'static str>
where
    P: FnOnce() -> Fut,
    Fut: std::future::Future<Output = ProbeOutcome>,
{
    let outcome = probe().await;
    let previous = live.get();
    let choice = select(outcome, previous, have_token).inspect_err(|_| {
        if outcome == ProbeOutcome::Rejected {
            live.observe(&async_nats::Event::ClientError(
                async_nats::ClientError::Other(
                    async_nats::ConnectErrorKind::AuthorizationViolation.to_string(),
                ),
            ));
        } else {
            *live.refusals.lock().unwrap_or_else(|e| e.into_inner()) = None;
            debug!(
                role = role.as_str(),
                "NATS credential probe inconclusive; deferring this attempt"
            );
        }
    })?;
    if outcome != ProbeOutcome::Unreachable {
        live.set(choice);
        live_for(role).set(choice);
    }
    // Log what was selected and the outcome class, never a value; quiet when
    // nothing changed so a flapping broker does not fill the log.
    if previous == Some(choice) {
        debug!(
            role = role.as_str(),
            credential = choice.label(),
            probe = outcome.label(),
            "NATS credential selected"
        );
    } else {
        info!(
            role = role.as_str(),
            credential = choice.label(),
            probe = outcome.label(),
            "NATS credential selected"
        );
    }
    Ok(choice)
}

/// Whether a connection needs per-attempt selection at all.
enum AuthPlan {
    /// No user: present the token (or nothing), exactly as before roles had
    /// users. No callback, no probe.
    Static(Option<String>),
    /// A user is provisioned: select per attempt.
    Select {
        token: Option<String>,
        user: UserCredential,
    },
}

fn auth_plan(creds: NatsCredentials) -> AuthPlan {
    match creds.user {
        None => AuthPlan::Static(creds.token),
        Some(user) => AuthPlan::Select {
            token: creds.token,
            user,
        },
    }
}

/// The `name` a kanade process announces on its NATS connection.
///
/// Without an identity this is `kanade-<role>`; with one it is
/// `kanade-<role>/<identity>`. The broker echoes it back verbatim in
/// `/connz`, which is what lets the backend attribute a connection — and
/// therefore the credential the broker authenticated it with — to a pc_id
/// (#1270). Nothing else on a connection carries the pc_id: the CID is
/// assigned by the server and the IP is not a stable identifier on a fleet
/// of laptops.
///
/// The name is client-supplied and therefore claimed, not proved. What
/// `/connz` makes unforgeable is the *credential* half of the pair; a host
/// can still lie about which pc_id it is. Under one fleet-wide token that
/// changes nothing (every host can already impersonate every other), and
/// closing it for good is per-agent identity, not a naming convention.
pub fn client_name(role: NatsRole, identity: Option<&str>) -> String {
    match identity.map(str::trim).filter(|s| !s.is_empty()) {
        Some(id) => format!("{NAME_PREFIX}{}{NAME_SEP}{id}", role.as_str()),
        None => format!("{NAME_PREFIX}{}", role.as_str()),
    }
}

/// A connection name split back into its parts — see [`client_name`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ClientName<'a> {
    /// The role segment as announced. A `&str` rather than a [`NatsRole`]
    /// on purpose: a connection from a future (or foreign) build may name a
    /// role this binary does not know, and dropping it on the floor would
    /// hide exactly the host worth looking at.
    pub role: &'a str,
    /// The host identity, when the connection carried one. `None` for the
    /// backend / CLI (which are not per-host) and for agents predating
    /// #1270 — those simply cannot be attributed.
    pub identity: Option<&'a str>,
}

/// Parse a connection name produced by [`client_name`]. `None` for any name
/// that is not a kanade connection at all (a `nats` CLI session, a
/// monitoring tool), which the caller should ignore rather than guess about.
pub fn parse_client_name(name: &str) -> Option<ClientName<'_>> {
    let rest = name.strip_prefix(NAME_PREFIX)?;
    Some(match rest.split_once(NAME_SEP) {
        // An empty identity (`kanade-agent/`) is not an identity.
        Some((role, id)) if !role.is_empty() && !id.is_empty() => ClientName {
            role,
            identity: Some(id),
        },
        Some((role, _)) if !role.is_empty() => ClientName {
            role,
            identity: None,
        },
        Some(_) => return None,
        None if !rest.is_empty() => ClientName {
            role: rest,
            identity: None,
        },
        None => return None,
    })
}

/// Answers "is this credential the one *we* present?" without handing the
/// credential itself to the caller.
///
/// #1270: the NATS monitoring endpoint reports `authorized_user` per
/// connection. A current nats-server hides that field for
/// token-authenticated connections, but that is the broker build's
/// behaviour, not a guarantee this side can lean on — a consumer of
/// `/connz` has to treat the value as possibly being the fleet-wide secret.
/// The one question it may safely answer about it is whether it equals the
/// credential this process already holds, and that answer is enough to
/// label a connection ("still on the shared token") without ever storing or
/// serving the value.
///
/// Constructed once and reused: [`resolve_token`] hits the Windows registry,
/// and the caller compares against every connection on the broker.
pub struct CredentialProbe {
    token: Option<String>,
    user: Option<String>,
    /// The connection's own decision, when this probe belongs to one. `None`
    /// for the explicit constructors, which describe a fixed credential.
    live: Option<Arc<Live>>,
}

/// Which shape of credential a [`CredentialProbe`] holds.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CredentialKind {
    None,
    Token,
    /// A named user — and, when the connection presenting it is **live**,
    /// positive proof that the broker is running `users` rather than a
    /// token: nats-server refuses to load a config carrying both a `token`
    /// and a `users` array ("Can not have a token and a users array") and
    /// rejects token authentication outright once `users` are defined.
    ///
    /// That proof is the only thing that makes a reported `authorized_user`
    /// safe to record verbatim. Note what is *not* proof: holding no
    /// credential locally. A process that never authenticated at all can
    /// still read a monitoring endpoint, and inferring the broker's mode
    /// from a local absence would let a misconfigured host store the very
    /// secret the rest of this type exists to protect.
    User,
}

/// Hand-written so a stray `{:?}` in a log line cannot print the credential.
/// A username is not a secret and is shown; a token never is.
impl std::fmt::Debug for CredentialProbe {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut held = Vec::new();
        if self.token.is_some() {
            held.push("<redacted token>".to_string());
        }
        if let Some(name) = &self.user {
            held.push(format!("user {name}"));
        }
        let rendered = if held.is_empty() {
            "<none>".to_string()
        } else {
            held.join(" + ")
        };
        f.debug_struct("CredentialProbe")
            .field("presented", &rendered)
            .field("kind", &self.kind())
            .finish()
    }
}

impl CredentialProbe {
    /// Resolve the credential `role` would present, exactly as [`connect`]
    /// does, and follow the decision that role's open connection makes.
    ///
    /// A half-provisioned user pair is ignored here — [`connect`] is where
    /// it fails loudly.
    pub fn for_role(role: NatsRole) -> Self {
        Self {
            token: resolve_token(role),
            user: resolve_user(role).ok().flatten().map(|u| u.name),
            live: Some(live_for(role)),
        }
    }

    /// Build a probe around an explicitly-supplied token.
    ///
    /// The production path is [`Self::for_role`]; this exists so callers can
    /// be tested against a known credential without a Windows registry to
    /// write to — and, on a developer's machine, without accidentally
    /// probing the real fleet token that `for_role` would find there.
    pub fn from_token(token: Option<String>) -> Self {
        Self {
            token,
            user: None,
            live: None,
        }
    }

    /// Build a probe for a process that authenticates as a named user.
    ///
    /// Describes a fixed user credential, for testing the consumers of
    /// [`CredentialKind::User`] without a connection.
    pub fn from_user(name: impl Into<String>) -> Self {
        Self {
            token: None,
            user: Some(name.into()),
            live: None,
        }
    }

    /// Which shape of credential this process is presenting **right now**.
    ///
    /// A role that holds a user but is on the token fallback reports
    /// [`CredentialKind::Token`]; the user shape is reported only once the
    /// broker has accepted it. Before any decision a role holding a token
    /// reports the token, and one holding only a user reports `None`: the
    /// user shape is proof of the broker's mode and is not claimed on a
    /// guess.
    pub fn kind(&self) -> CredentialKind {
        let decided = self.live.as_ref().and_then(|l| l.get());
        match decided {
            Some(Choice::User) if self.user.is_some() => return CredentialKind::User,
            Some(Choice::Token) if self.token.is_some() => return CredentialKind::Token,
            _ => {}
        }
        if self.live.is_none() && self.user.is_some() {
            CredentialKind::User
        } else if self.token.is_some() {
            CredentialKind::Token
        } else {
            CredentialKind::None
        }
    }

    /// Whether `candidate` is the **secret** this process presents.
    ///
    /// Only ever true for a token. A user's secret is its password, which
    /// `authorized_user` never carries — matching a username here would
    /// mean "this connection is on the same account", a different and much
    /// weaker statement than the one callers use this for.
    ///
    /// A plain comparison: `candidate` comes from the broker's own report of
    /// connections it already authenticated, not from an attacker-chosen
    /// input, so there is no oracle to time.
    pub fn is_ours(&self, candidate: &str) -> bool {
        self.token.as_deref() == Some(candidate)
    }
}

/// Connect to NATS at `url` as `role`. Resolves the credential from the
/// registry (Windows) or the environment: a user pair when one is
/// provisioned (with the token as fallback, see the module docs), otherwise
/// the bearer token; connects unauthenticated when neither is set.
///
/// The connection is announced as `kanade-<role>` with no host identity —
/// right for the backend and the CLI, which are not per-host. A role that
/// has to be attributable to a specific machine (the agent) must use
/// [`connect_with_event_callback`] and pass one; see [`client_name`].
pub async fn connect(role: NatsRole, url: &str) -> Result<async_nats::Client> {
    connect_inner(
        role,
        url,
        None,
        None::<fn(async_nats::Event) -> std::future::Ready<()>>,
        NatsCredentials::resolve(role)?,
        Arc::new(Live::default()),
    )
    .await
}

/// Same as [`connect`] but also wires an `event_callback` that fires
/// whenever async-nats publishes a `ConnectEvent` (Connected,
/// Disconnected, ServerError, etc.). The callback's `Future` runs on
/// the async-nats internal task — keep it cheap and non-blocking
/// (set a flag, send on a channel, that kind of thing) so the
/// connection state machine isn't held up.
///
/// Used by the agent's v0.26 Layer 2 staleness tracker: the callback
/// stamps a shared `Mutex<Option<Instant>>` on every Connected event,
/// so `decide()` at fire time can answer "how long ago were we last
/// definitely-talking-to-the-broker" without a polling loop.
///
/// `identity` names the host this connection belongs to (the agent's
/// pc_id). It becomes part of the connection name the broker echoes in
/// `/connz`, which is the only thing tying a connection — and the
/// credential that opened it — back to a machine (#1270).
pub async fn connect_with_event_callback<F, Fut>(
    role: NatsRole,
    url: &str,
    identity: Option<&str>,
    cb: F,
) -> Result<async_nats::Client>
where
    F: Fn(async_nats::Event) -> Fut + Send + Sync + 'static,
    Fut: std::future::Future<Output = ()> + Send + Sync + 'static,
{
    connect_inner(
        role,
        url,
        identity,
        Some(cb),
        NatsCredentials::resolve(role)?,
        Arc::new(Live::default()),
    )
    .await
}

/// Connect with explicitly supplied credentials instead of resolving them
/// from the registry / environment, with refusal state private to this
/// connection. Credential reporting follows the latest decision for the role.
pub async fn connect_with_credentials(
    role: NatsRole,
    url: &str,
    creds: NatsCredentials,
) -> Result<async_nats::Client> {
    connect_inner(
        role,
        url,
        None,
        None::<fn(async_nats::Event) -> std::future::Ready<()>>,
        creds,
        Arc::new(Live::default()),
    )
    .await
}

/// [`connect_with_credentials`] with an event callback, so a test can observe
/// what the connection reports while it cannot authenticate.
pub async fn connect_with_credentials_and_event_callback<F, Fut>(
    role: NatsRole,
    url: &str,
    creds: NatsCredentials,
    cb: F,
) -> Result<async_nats::Client>
where
    F: Fn(async_nats::Event) -> Fut + Send + Sync + 'static,
    Fut: std::future::Future<Output = ()> + Send + Sync + 'static,
{
    connect_inner(role, url, None, Some(cb), creds, Arc::new(Live::default())).await
}

async fn connect_inner<F, Fut>(
    role: NatsRole,
    url: &str,
    identity: Option<&str>,
    cb: Option<F>,
    creds: NatsCredentials,
    live: Arc<Live>,
) -> Result<async_nats::Client>
where
    F: Fn(async_nats::Event) -> Fut + Send + Sync + 'static,
    Fut: std::future::Future<Output = ()> + Send + Sync + 'static,
{
    // #1187: a workspace build unifies rustls features, so any binary built
    // alongside a reqwest user (the CLI, the backend) links BOTH aws-lc-rs and
    // ring. rustls 0.23 then cannot auto-pick a process-level provider, and the
    // first TLS handshake — a `wss://`/`tls://` broker — panics inside
    // async-nats' connection task. That panic does not surface here: the
    // client just goes dead. Every production connect funnels through this
    // function, so installing ring here covers every binary. `install_default`
    // returns Err when a provider is already installed — ignore it.
    let _ = rustls::crypto::ring::default_provider().install_default();

    let health = Arc::new(ConnectionHealth {
        live: live.clone(),
        url: url.to_string(),
        creds: creds.clone(),
    });
    publish_role_live(role, &live);
    let tracker = live.clone();
    // Only a provisioned user changes how the credential is presented; every
    // other host takes the token path untouched.
    let opts = match auth_plan(creds) {
        AuthPlan::Static(token) => {
            let opts = async_nats::ConnectOptions::new();
            match token {
                Some(token) => opts.token(token),
                None => opts,
            }
        }
        AuthPlan::Select { token, user } => {
            let url = url.to_string();
            async_nats::ConnectOptions::with_auth_callback(move |_nonce| {
                let (url, token, user, live) =
                    (url.clone(), token.clone(), user.clone(), live.clone());
                async move {
                    let choice = choose(role, token.is_some(), &live, || probe_user(&url, &user))
                        .await
                        .map_err(async_nats::AuthError::new)?;
                    let mut auth = async_nats::Auth::new();
                    match choice {
                        Choice::User => {
                            auth.username = Some(user.name);
                            auth.password = Some(user.password);
                        }
                        Choice::Token => auth.token = token,
                    }
                    Ok(auth)
                }
            })
        }
    };

    // v0.38 / #137: offline-tolerant boot. Without
    // `retry_on_initial_connect`, `opts.connect(url).await` blocks-then-
    // errors when the broker is unreachable at startup — the agent
    // process dies, SCM ticks its restart counter, and the offline-
    // tolerant subsystems (local_scheduler, outbox drain) never spawn.
    // With this flag, connect() returns `Ok(Client)` immediately and
    // async-nats does the reconnect in the background; subscribe()
    // calls queue the SUB frame until the link is up.
    let opts = opts
        .retry_on_initial_connect()
        .connection_timeout(HEALTH_TIMEOUT)
        .ping_interval(PING_INTERVAL)
        // Names the connection in `nats server report connections`, in the
        // broker's own logs, and in `/connz`. Free observability while the
        // fleet is mid-migration: it shows which roles are connecting even
        // before their credentials differ, which is exactly the window in
        // which a wrongly-provisioned host is otherwise invisible. With an
        // identity it also carries the pc_id, so #1270 can join the broker's
        // per-connection `authorized_user` back onto the agents row.
        .name(client_name(role, identity));
    // Always installed, even without a caller callback: it is how a refused
    // credential is noticed, since async-nats only retries it.
    let opts = opts.event_callback(move |ev| {
        tracker.observe(&ev);
        let forwarded = cb.as_ref().map(|cb| cb(ev));
        async move {
            if let Some(forwarded) = forwarded {
                forwarded.await;
            }
        }
    });
    let client = opts
        .connect(url)
        .await
        .with_context(|| format!("connect to NATS at {url}"))?;
    register_health(&client, health);
    Ok(client)
}

/// Whether the client's connection task has terminated.
///
/// When that task ends (a panic inside it, a drain, exhausted reconnects) the
/// `Client` stays alive and never talks again — a silent zombie. The
/// observable difference from an ordinary disconnect is the command channel:
/// a task that is alive but reconnecting simply does not answer a flush, so
/// the flush waits and the timeout elapses; a task that is gone has dropped
/// the receiving end, so queueing the flush fails with a send error. The
/// check publishes to no subject, so no role needs any publish right for it.
///
/// This tells whether the task still accepts commands, not whether the broker
/// is reachable. A flush that fails after it was queued (its observer was
/// dropped) is not treated as death: a task that really is gone fails the
/// next check at the send step, and mistaking a plain disconnect for death
/// would make a process exit needlessly.
pub async fn is_dead(client: &async_nats::Client) -> bool {
    match tokio::time::timeout(Duration::from_secs(5), client.flush()).await {
        Ok(Err(e)) => match e.kind() {
            async_nats::client::FlushErrorKind::SendError => true,
            async_nats::client::FlushErrorKind::FlushError => false,
        },
        Ok(Ok(())) | Err(_) => false,
    }
}

/// Resolve once `client` can no longer be expected to work: its connection
/// task has terminated (see [`is_dead`]), its credential is persistently
/// refused, or receive progress stalls while a fresh handshake reaches the
/// broker (accepted or explicitly refused).
/// Protocol PING/PONG traffic provides progress even for idle roles. A flush
/// alone only proves a local write, not that the broker answered. The local
/// check runs every `liveness.check_interval`; a witness connection is made
/// only after `liveness.stall_bound` without received bytes, never on a
/// healthy connection, and an unreachable broker never ends the wait. A
/// process that depends on the client should treat this as fatal and exit
/// non-zero so its service manager restarts it.
///
/// `role` names the connection the client was opened with.
pub async fn wait_until_dead_with(role: NatsRole, client: &async_nats::Client, liveness: Liveness) {
    let health = connection_health(client);
    let mut tick = tokio::time::interval(liveness.check_interval);
    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    let stats = client.statistics();
    let mut received = stats.in_bytes.load(Ordering::Relaxed);
    let mut judge = StallJudge::new(liveness, tokio::time::Instant::now());
    loop {
        tick.tick().await;
        let now_received = stats.in_bytes.load(Ordering::Relaxed);
        if client.connection_state() == async_nats::connection::State::Connected
            && now_received != received
        {
            judge.progressed(tokio::time::Instant::now());
        }
        received = now_received;
        if health.as_ref().is_some_and(|h| h.live.auth_failed()) {
            warn!(
                role = role.as_str(),
                "the NATS broker keeps refusing this role's credential; the client cannot connect"
            );
            return;
        }
        if is_dead(client).await {
            warn!(
                role = role.as_str(),
                "NATS connection task has terminated; the client can no longer talk to the broker"
            );
            return;
        }
        let Some(health) = &health else { continue };
        if !judge.witness_due(tokio::time::Instant::now()) {
            continue;
        }
        let outcome = witness(health).await;
        if judge.witnessed(tokio::time::Instant::now(), outcome) != Verdict::Stalled {
            continue;
        }
        // Recheck after the witness: a reconnect racing it is healthy.
        if client.connection_state() == async_nats::connection::State::Connected
            && stats.in_bytes.load(Ordering::Relaxed) != received
        {
            judge.progressed(tokio::time::Instant::now());
            continue;
        }
        warn!(
            role = role.as_str(),
            witness = outcome.label(),
            "NATS client made no receive progress while a fresh credential witness reached the broker; exiting for a supervised restart"
        );
        return;
    }
}

/// [`wait_until_dead_with`] at the production stall bound, checking every
/// `interval`.
pub async fn wait_until_dead_every(
    role: NatsRole,
    client: &async_nats::Client,
    interval: Duration,
) {
    wait_until_dead_with(
        role,
        client,
        Liveness {
            check_interval: interval,
            ..Liveness::PRODUCTION
        },
    )
    .await
}

/// [`wait_until_dead_with`] at the production timing.
pub async fn wait_until_dead(role: NatsRole, client: &async_nats::Client) {
    wait_until_dead_with(role, client, Liveness::PRODUCTION).await
}

/// Start supervision at `liveness` and exit the process when it resolves.
pub fn exit_on_dead_with(role: NatsRole, client: &async_nats::Client, liveness: Liveness) {
    let client = client.clone();
    tokio::spawn(async move {
        wait_until_dead_with(role, &client, liveness).await;
        std::process::exit(1);
    });
}

/// Start supervision before subscriptions or bootstrap can block. Fatal
/// liveness failure bypasses application shutdown so a blocked subsystem
/// cannot keep a silent service alive beyond the exit deadline.
pub fn exit_on_dead(role: NatsRole, client: &async_nats::Client) {
    exit_on_dead_with(role, client, Liveness::PRODUCTION);
}

#[cfg(test)]
mod tests {
    use super::*;

    fn secs(n: u64) -> Duration {
        Duration::from_secs(n)
    }

    /// Drive `judge` on the local check cadence until `until`, with receive
    /// progress every `progress_every` (none: silent). Returns the first time
    /// a witness was due and, if it ends in a stall verdict, when.
    fn simulate(
        judge: &mut StallJudge,
        start: tokio::time::Instant,
        until: Duration,
        progress_every: Option<Duration>,
        outcome: ProbeOutcome,
        witnesses: &mut Vec<Duration>,
    ) -> Option<Duration> {
        let step = judge.liveness.check_interval;
        let mut elapsed = Duration::ZERO;
        let mut last_progress = Duration::ZERO;
        while elapsed < until {
            elapsed += step;
            let now = start + elapsed;
            if progress_every.is_some_and(|p| elapsed - last_progress >= p) {
                last_progress = elapsed;
                judge.progressed(now);
            }
            if judge.witness_due(now) {
                witnesses.push(elapsed);
                if judge.witnessed(now, outcome) == Verdict::Stalled {
                    return Some(elapsed);
                }
            }
        }
        None
    }

    #[test]
    fn the_production_bounds_leave_a_healthy_idle_connection_a_wide_margin() {
        assert!(STALL_BOUND >= PING_INTERVAL + secs(60));
        assert!(EXIT_BOUND < secs(300));
        assert_eq!(PING_INTERVAL, secs(60));
        assert_eq!(STALL_BOUND, secs(180));
        assert_eq!(EXIT_BOUND, secs(195));
    }

    #[test]
    fn an_idle_healthy_client_is_never_judged_stalled() {
        let start = tokio::time::Instant::now();
        let mut judge = StallJudge::new(Liveness::PRODUCTION, start);
        let mut witnesses = Vec::new();
        // One PONG per ping interval, for a day.
        let verdict = simulate(
            &mut judge,
            start,
            secs(24 * 3600),
            Some(PING_INTERVAL),
            ProbeOutcome::Accepted,
            &mut witnesses,
        );
        assert_eq!(verdict, None);
        assert!(
            witnesses.is_empty(),
            "a witness connected to a healthy broker"
        );
    }

    #[test]
    fn an_offline_broker_never_exits_and_is_polled_at_the_resume_interval() {
        let start = tokio::time::Instant::now();
        let mut judge = StallJudge::new(Liveness::PRODUCTION, start);
        let mut witnesses = Vec::new();
        let verdict = simulate(
            &mut judge,
            start,
            secs(3 * 3600),
            None,
            ProbeOutcome::Unreachable,
            &mut witnesses,
        );
        assert_eq!(verdict, None);
        assert_eq!(witnesses[0], STALL_BOUND);
        assert!(witnesses.windows(2).all(|w| w[1] - w[0] >= RESUME_BOUND));
    }

    #[test]
    fn a_stall_after_an_outage_is_judged_against_the_normal_bound() {
        let start = tokio::time::Instant::now();
        let mut judge = StallJudge::new(Liveness::PRODUCTION, start);
        let mut witnesses = Vec::new();
        assert_eq!(
            simulate(
                &mut judge,
                start,
                secs(3600),
                None,
                ProbeOutcome::Unreachable,
                &mut witnesses
            ),
            None
        );
        assert!(judge.outage);
        // The broker answers but the client has not resumed: only a grace.
        let back = start + secs(3600);
        witnesses.clear();
        let exit = simulate(
            &mut judge,
            back,
            EXIT_BOUND,
            None,
            ProbeOutcome::Accepted,
            &mut witnesses,
        );
        assert!(!judge.outage);
        let exit = exit.expect("a silent client with a reachable broker must exit");
        assert!(exit <= EXIT_BOUND, "{exit:?}");

        // Progress after the outage clears it; a later genuine stall is then
        // judged against the full stall bound from the last progress.
        let mut judge = StallJudge::new(Liveness::PRODUCTION, start);
        let mut witnesses = Vec::new();
        simulate(
            &mut judge,
            start,
            secs(3600),
            None,
            ProbeOutcome::Unreachable,
            &mut witnesses,
        );
        assert!(judge.outage);
        let resumed = start + secs(3600);
        judge.progressed(resumed);
        assert!(!judge.outage);
        assert!(!judge.witness_due(resumed + STALL_BOUND - secs(1)));
        witnesses.clear();
        let exit = simulate(
            &mut judge,
            resumed,
            STALL_BOUND + secs(60),
            None,
            ProbeOutcome::Rejected,
            &mut witnesses,
        );
        assert_eq!(exit, Some(STALL_BOUND));
    }
    use std::collections::HashMap;

    /// A stand-in registry. Keys are `subkey\value`.
    fn reg(entries: &[(&str, &str)]) -> impl Fn(&str, &str) -> Option<String> {
        let map: HashMap<String, String> = entries
            .iter()
            .map(|(k, v)| ((*k).to_string(), (*v).to_string()))
            .collect();
        move |subkey: &str, value: &str| map.get(&format!(r"{subkey}\{value}")).cloned()
    }

    #[test]
    fn role_subkeys_are_distinct_and_agent_matches_the_shared_path() {
        assert_eq!(NatsRole::Backend.reg_subkey(), r"SOFTWARE\kanade\backend");
        assert_eq!(NatsRole::Cli.reg_subkey(), r"SOFTWARE\kanade\cli");
        // The agent's role key IS the historical shared key, so an agent
        // never sees a migration at all.
        assert_eq!(NatsRole::Agent.reg_subkey(), REG_SHARED_SUBKEY);
    }

    #[test]
    fn an_unmigrated_fleet_keeps_presenting_the_shared_token() {
        // The state of every host today: one token, under the agent path.
        let registry = reg(&[(r"SOFTWARE\kanade\agent\NatsToken", "shared")]);
        for role in [NatsRole::Agent, NatsRole::Backend, NatsRole::Cli] {
            assert_eq!(
                resolve_token_with(role, &registry, None).as_deref(),
                Some("shared"),
                "{role:?} must keep working before its own key is provisioned"
            );
        }
    }

    #[test]
    fn a_role_key_wins_over_the_shared_one() {
        let registry = reg(&[
            (r"SOFTWARE\kanade\agent\NatsToken", "shared"),
            (r"SOFTWARE\kanade\backend\NatsToken", "backend-only"),
        ]);
        // The migrated role uses its own credential...
        assert_eq!(
            resolve_token_with(NatsRole::Backend, &registry, None).as_deref(),
            Some("backend-only")
        );
        // ...while a role that has not been migrated yet is unaffected.
        assert_eq!(
            resolve_token_with(NatsRole::Cli, &registry, None).as_deref(),
            Some("shared")
        );
    }

    #[test]
    fn removing_a_role_key_falls_back_rather_than_failing() {
        // Rollback of a per-host migration step: the role key is gone, and
        // the host must return to the shared credential instead of
        // connecting unauthenticated (which a broker with `authorization`
        // would refuse — turning a rollback into an outage).
        let registry = reg(&[(r"SOFTWARE\kanade\agent\NatsToken", "shared")]);
        assert_eq!(
            resolve_token_with(NatsRole::Backend, &registry, None).as_deref(),
            Some("shared")
        );
    }

    #[test]
    fn the_registry_outranks_the_environment() {
        // Unchanged from before roles existed: a dev shell's env var must
        // not quietly override a provisioned production credential.
        let registry = reg(&[(r"SOFTWARE\kanade\agent\NatsToken", "shared")]);
        assert_eq!(
            resolve_token_with(NatsRole::Agent, &registry, Some("from-env".into())).as_deref(),
            Some("shared")
        );
    }

    #[test]
    fn the_environment_serves_hosts_with_no_registry_at_all() {
        let empty = reg(&[]);
        assert_eq!(
            resolve_token_with(NatsRole::Cli, &empty, Some("from-env".into())).as_deref(),
            Some("from-env")
        );
        // An empty env var is not a credential — it must fall through to
        // "no token" so a dev broker without `authorization` still works,
        // rather than presenting the empty string and being rejected.
        assert_eq!(
            resolve_token_with(NatsRole::Cli, &empty, Some(String::new())),
            None
        );
        assert_eq!(resolve_token_with(NatsRole::Cli, &empty, None), None);
    }

    // ── #1270: connection naming ─────────────────────────────────────

    #[test]
    fn an_identity_round_trips_through_the_connection_name() {
        // The pc_id is the join key between `/connz` and the agents table,
        // so the name has to survive the trip unchanged — including the
        // casing, which is NOT uniform across the fleet and which NATS
        // subjects treat as significant.
        for pc in ["PC001", "minipc", "Web%01", "ws-9"] {
            let name = client_name(NatsRole::Agent, Some(pc));
            let parsed = parse_client_name(&name).expect("our own name must parse");
            assert_eq!(parsed.role, "agent");
            assert_eq!(parsed.identity, Some(pc));
        }
    }

    #[test]
    fn a_role_without_an_identity_keeps_the_pre_1270_name() {
        // The backend and the CLI are not per-host, and an agent that
        // predates #1270 announces this shape too. Both must parse as "a
        // kanade connection we cannot attribute" rather than as an error or
        // as an empty pc_id.
        assert_eq!(client_name(NatsRole::Backend, None), "kanade-backend");
        let parsed = parse_client_name("kanade-agent").unwrap();
        assert_eq!(parsed.role, "agent");
        assert_eq!(parsed.identity, None);
        // Whitespace-only is not an identity either — it would otherwise
        // produce a name that parses back into a pc_id no row can match.
        assert_eq!(client_name(NatsRole::Agent, Some("  ")), "kanade-agent");
    }

    #[test]
    fn foreign_connections_do_not_parse_as_kanade_ones() {
        // A `nats` CLI session or a monitoring tool shares the broker. The
        // projector must skip those rather than attribute them to a host.
        assert!(parse_client_name("NATS CLI Version 0.1.5").is_none());
        assert!(parse_client_name("").is_none());
        assert!(parse_client_name("kanade-").is_none());
        assert!(parse_client_name("kanade-/PC001").is_none());
        // A trailing separator with no identity is a role, not a pc_id.
        assert_eq!(parse_client_name("kanade-agent/").unwrap().identity, None);
    }

    #[test]
    fn an_unknown_role_is_preserved_rather_than_dropped() {
        // A future build (or something impersonating one) naming a role this
        // binary has never heard of is precisely the connection an operator
        // wants to see.
        let parsed = parse_client_name("kanade-relay/PC001").unwrap();
        assert_eq!(parsed.role, "relay");
        assert_eq!(parsed.identity, Some("PC001"));
    }

    // ── #1270: credential probe ──────────────────────────────────────

    #[test]
    fn the_probe_recognises_only_the_credential_we_present() {
        let probe = CredentialProbe::from_token(Some("shared".into()));
        assert_eq!(probe.kind(), CredentialKind::Token);
        assert!(probe.is_ours("shared"));
        assert!(!probe.is_ours("something-else"));
        // The empty string is what the broker reports for a connection it
        // did not authenticate at all. It must never read as "ours".
        assert!(!probe.is_ours(""));
    }

    #[test]
    fn a_probe_with_no_credential_matches_nothing() {
        // Dev broker with no `authorization` block. We hold nothing, so we
        // can prove nothing about anyone else's credential — including that
        // it is safe to store.
        let probe = CredentialProbe::from_token(None);
        assert_eq!(probe.kind(), CredentialKind::None);
        assert!(!probe.is_ours(""));
        assert!(!probe.is_ours("anything"));
    }

    #[test]
    fn a_username_is_not_a_secret_we_can_recognise() {
        // `is_ours` answers "is this MY secret", and a user's secret is its
        // password. Matching on the username instead would answer a much
        // weaker question while reading like the strong one.
        let probe = CredentialProbe::from_user("kanade-backend");
        assert_eq!(probe.kind(), CredentialKind::User);
        assert!(!probe.is_ours("kanade-backend"));
    }

    #[test]
    fn the_probe_never_prints_the_credential() {
        // `/connz` handling logs liberally; one `{:?}` on the probe must not
        // be the thing that puts the fleet's token in a log file.
        let probe = CredentialProbe::from_token(Some("super-secret-token".into()));
        let rendered = format!("{probe:?}");
        assert!(!rendered.contains("super-secret-token"), "{rendered}");
        assert!(rendered.contains("redacted"), "{rendered}");
        assert!(
            format!("{:?}", CredentialProbe::from_token(None)).contains("none"),
            "the no-credential case should be visible, just not the value"
        );
        // A username is not a secret — hiding it would cost diagnosability
        // for nothing.
        assert!(
            format!("{:?}", CredentialProbe::from_user("kanade-backend"))
                .contains("kanade-backend"),
        );
    }

    // ── per-role users ───────────────────────────────────────────────

    fn user(r: Result<Option<UserCredential>>) -> Option<(String, String)> {
        r.unwrap().map(|u| (u.name, u.password))
    }

    #[test]
    fn a_users_registry_pair_comes_from_the_roles_own_key() {
        let registry = reg(&[
            (r"SOFTWARE\kanade\backend\NatsUser", "kanade-backend"),
            (r"SOFTWARE\kanade\backend\NatsPassword", "pw"),
        ]);
        assert_eq!(
            user(resolve_user_with(NatsRole::Backend, &registry, None, None)),
            Some(("kanade-backend".into(), "pw".into()))
        );
        // No shared user: another role does not inherit it, not even the
        // agent's key (which is where the shared *token* lives).
        for role in [NatsRole::Agent, NatsRole::Cli] {
            assert_eq!(user(resolve_user_with(role, &registry, None, None)), None);
        }
        let agent = reg(&[
            (r"SOFTWARE\kanade\agent\NatsUser", "a"),
            (r"SOFTWARE\kanade\agent\NatsPassword", "b"),
        ]);
        assert_eq!(
            user(resolve_user_with(NatsRole::Cli, &agent, None, None)),
            None
        );
    }

    #[test]
    fn the_registry_pair_outranks_the_environment_pair() {
        let registry = reg(&[
            (r"SOFTWARE\kanade\cli\NatsUser", "reg-user"),
            (r"SOFTWARE\kanade\cli\NatsPassword", "reg-pw"),
        ]);
        assert_eq!(
            user(resolve_user_with(
                NatsRole::Cli,
                &registry,
                Some("env-user".into()),
                Some("env-pw".into())
            )),
            Some(("reg-user".into(), "reg-pw".into()))
        );
    }

    #[test]
    fn the_environment_serves_a_user_pair_when_the_registry_has_none() {
        assert_eq!(
            user(resolve_user_with(
                NatsRole::Cli,
                reg(&[]),
                Some("u".into()),
                Some("p".into())
            )),
            Some(("u".into(), "p".into()))
        );
        // Empty values are not a credential.
        assert_eq!(
            user(resolve_user_with(
                NatsRole::Cli,
                reg(&[]),
                Some(String::new()),
                Some(String::new())
            )),
            None
        );
        assert_eq!(
            user(resolve_user_with(NatsRole::Cli, reg(&[]), None, None)),
            None
        );
    }

    #[test]
    fn half_a_pair_is_an_error_not_a_fallback() {
        let err = |r: Result<Option<UserCredential>>| format!("{:#}", r.unwrap_err());
        // Registry halves.
        let only_user = reg(&[(r"SOFTWARE\kanade\cli\NatsUser", "u")]);
        assert!(
            err(resolve_user_with(NatsRole::Cli, &only_user, None, None))
                .contains("password is missing")
        );
        let only_pw = reg(&[(r"SOFTWARE\kanade\cli\NatsPassword", "secret-pw")]);
        let e = err(resolve_user_with(NatsRole::Cli, &only_pw, None, None));
        assert!(e.contains("user name is missing"), "{e}");
        assert!(!e.contains("secret-pw"), "{e}");
        // A half registry pair is not rescued by a complete environment pair.
        assert!(
            resolve_user_with(
                NatsRole::Cli,
                &only_user,
                Some("u".into()),
                Some("p".into())
            )
            .is_err()
        );
        // Environment halves.
        assert!(resolve_user_with(NatsRole::Cli, reg(&[]), Some("u".into()), None).is_err());
        assert!(resolve_user_with(NatsRole::Cli, reg(&[]), None, Some("p".into())).is_err());
    }

    #[test]
    fn credentials_never_print_a_secret() {
        let creds = NatsCredentials::new(
            Some("tok-secret".into()),
            Some(("kanade-agent".into(), "pw-secret".into())),
        );
        let rendered = format!("{creds:?}");
        assert!(!rendered.contains("tok-secret"), "{rendered}");
        assert!(!rendered.contains("pw-secret"), "{rendered}");
        assert!(rendered.contains("kanade-agent"), "{rendered}");
    }

    #[test]
    fn no_user_means_the_static_token_path_with_no_selection() {
        match auth_plan(NatsCredentials::new(Some("t".into()), None)) {
            AuthPlan::Static(Some(t)) => assert_eq!(t, "t"),
            _ => panic!("a token-only host must take the static path"),
        }
        assert!(matches!(
            auth_plan(NatsCredentials::new(None, None)),
            AuthPlan::Static(None)
        ));
        assert!(matches!(
            auth_plan(NatsCredentials::new(
                Some("t".into()),
                Some(("u".into(), "p".into()))
            )),
            AuthPlan::Select { .. }
        ));
    }

    #[test]
    fn selection_follows_the_probe() {
        use Choice::{Token, User};
        use ProbeOutcome::*;
        assert_eq!(select(Accepted, None, true), Ok(User));
        assert_eq!(select(Accepted, Some(Token), false), Ok(User));
        assert_eq!(select(Rejected, Some(User), true), Ok(Token));
        let missing = select(Rejected, None, false).unwrap_err();
        assert!(missing.contains("NatsToken"), "{missing}");
        // An inconclusive probe defers even when a previous mode is cached.
        assert!(select(Unreachable, None, true).is_err());
        assert!(select(Unreachable, Some(Token), true).is_err());
        assert!(select(Unreachable, Some(User), true).is_err());
    }

    #[tokio::test]
    async fn only_a_probe_answer_is_remembered() {
        let live = Live::default();
        let role = NatsRole::Cli;
        let run = |o| choose(role, true, &live, move || async move { o });
        // An unreachable broker cannot justify a handshake.
        assert!(run(ProbeOutcome::Unreachable).await.is_err());
        assert_eq!(live.get(), None);
        assert_eq!(run(ProbeOutcome::Rejected).await, Ok(Choice::Token));
        assert_eq!(live.get(), Some(Choice::Token));
        // The broker goes away: its previous mode is no longer evidence.
        assert!(run(ProbeOutcome::Unreachable).await.is_err());
        // ...and the flip back is followed.
        assert_eq!(run(ProbeOutcome::Accepted).await, Ok(Choice::User));
        assert_eq!(live.get(), Some(Choice::User));
        // A rejection with no token fails and leaves the cache alone.
        let err = choose(role, false, &live, || async { ProbeOutcome::Rejected })
            .await
            .unwrap_err();
        assert!(err.contains("NatsToken"));
        assert_eq!(live.get(), Some(Choice::User));
    }

    #[test]
    fn the_probe_reports_the_shape_in_use_right_now() {
        let live = Arc::new(Live::default());
        let probe = CredentialProbe {
            token: Some("tok".into()),
            user: Some("kanade-agent".into()),
            live: Some(live.clone()),
        };
        // Nothing proven yet: not the user shape.
        assert_eq!(probe.kind(), CredentialKind::Token);
        live.set(Choice::User);
        assert_eq!(probe.kind(), CredentialKind::User);
        // Holding a user while on the token fallback is the token shape.
        live.set(Choice::Token);
        assert_eq!(probe.kind(), CredentialKind::Token);
        assert!(probe.is_ours("tok"));

        let user_only = CredentialProbe {
            token: None,
            user: Some("u".into()),
            live: Some(Arc::new(Live::default())),
        };
        assert_eq!(user_only.kind(), CredentialKind::None);
        user_only.live.as_ref().unwrap().set(Choice::User);
        assert_eq!(user_only.kind(), CredentialKind::User);
    }

    #[test]
    fn a_probe_holding_both_credentials_prints_neither_secret() {
        let probe = CredentialProbe {
            token: Some("tok-secret".into()),
            user: Some("kanade-agent".into()),
            live: None,
        };
        let rendered = format!("{probe:?}");
        assert!(!rendered.contains("tok-secret"), "{rendered}");
        assert!(rendered.contains("kanade-agent"), "{rendered}");
    }

    #[test]
    fn sustained_refusals_fail_the_client_and_a_connect_clears_them() {
        let refused = || {
            async_nats::Event::ClientError(async_nats::ClientError::Other(
                async_nats::ConnectErrorKind::AuthorizationViolation.to_string(),
            ))
        };
        let live = Live::default();
        for _ in 0..AUTH_REJECTION_LIMIT {
            live.observe(&refused());
        }
        // Enough refusals, but not over a long enough window: a broker
        // mid-reload must not count.
        assert!(!live.auth_failed());
        // Age the first refusal past the window.
        let aged = std::time::Instant::now() - AUTH_REJECTION_WINDOW - Duration::from_secs(1);
        *live.refusals.lock().unwrap() = Some((aged, AUTH_REJECTION_LIMIT));
        assert!(live.auth_failed());
        live.observe(&async_nats::Event::Connected);
        assert!(!live.auth_failed());
        // A different kind of failure (broker gone) ends the run, so refusals
        // from before an outage cannot add up to a failure after it.
        *live.refusals.lock().unwrap() = Some((aged, AUTH_REJECTION_LIMIT));
        live.observe(&async_nats::Event::ClientError(
            async_nats::ClientError::Other("io".into()),
        ));
        assert!(!live.auth_failed());
        live.observe(&refused());
        assert!(!live.auth_failed());
    }

    #[tokio::test]
    async fn deferred_probes_do_not_count_as_credential_refusals() {
        let live = Live::default();
        for _ in 0..AUTH_REJECTION_LIMIT {
            assert!(
                choose(NatsRole::Cli, true, &live, || async {
                    ProbeOutcome::Unreachable
                })
                .await
                .is_err()
            );
            live.observe(&async_nats::Event::ClientError(
                async_nats::ClientError::Other(
                    async_nats::ConnectErrorKind::Authentication.to_string(),
                ),
            ));
        }
        assert!(live.refusals.lock().unwrap().is_none());
        for _ in 0..AUTH_REJECTION_LIMIT {
            assert!(
                choose(NatsRole::Cli, false, &live, || async {
                    ProbeOutcome::Rejected
                })
                .await
                .is_err()
            );
            live.observe(&async_nats::Event::ClientError(
                async_nats::ClientError::Other(
                    async_nats::ConnectErrorKind::Authentication.to_string(),
                ),
            ));
        }
        assert_eq!(
            live.refusals.lock().unwrap().as_ref().unwrap().1,
            AUTH_REJECTION_LIMIT
        );
        assert!(!live.auth_failed(), "quick refusals must still allow retry");
        assert!(
            choose(NatsRole::Cli, true, &live, || async {
                ProbeOutcome::Unreachable
            })
            .await
            .is_err()
        );
        assert!(
            live.refusals.lock().unwrap().is_none(),
            "an outage interrupts the refusal run"
        );
    }

    #[tokio::test]
    async fn same_role_connections_have_independent_health_state() {
        let first = connect_with_credentials(
            NatsRole::Agent,
            "nats://127.0.0.1:1",
            NatsCredentials::default(),
        )
        .await
        .unwrap();
        let second = connect_with_credentials(
            NatsRole::Agent,
            "nats://127.0.0.1:1",
            NatsCredentials::default(),
        )
        .await
        .unwrap();
        let first_health = connection_health(&first).unwrap();
        let second_health = connection_health(&second).unwrap();
        assert!(!Arc::ptr_eq(&first_health.live, &second_health.live));
        assert!(Arc::ptr_eq(
            &first_health,
            &connection_health(&first.clone()).unwrap()
        ));
        *first_health.live.refusals.lock().unwrap() = Some((
            std::time::Instant::now() - AUTH_REJECTION_WINDOW,
            AUTH_REJECTION_LIMIT,
        ));
        assert!(first_health.live.auth_failed());
        assert!(!second_health.live.auth_failed());
    }

    #[test]
    fn connect_leaves_a_process_crypto_provider_installed() {
        // #1187: a binary that links both aws-lc-rs and ring (any workspace
        // build) panics on its first wss handshake unless a process-level
        // provider was installed first. That panic happens inside async-nats'
        // background task, so the only thing a caller ever sees is a client
        // that silently never works — pin the precondition here instead.
        // `retry_on_initial_connect` makes connect return without a broker.
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(async {
            connect(NatsRole::Agent, "nats://127.0.0.1:1")
                .await
                .unwrap();
        });
        assert!(rustls::crypto::CryptoProvider::get_default().is_some());
    }
}