node-app-build 6.11.0

//! Fail-closed operation-mode coordinator for `node-app dev --client-node
//! --operation-mode`.
//!
//! When a developer opens the client-node PWA in a fresh browser, that browser
//! registers itself as a *pending* device on the owner's DID. This coordinator
//! automates the one manual step that would otherwise block the operational
//! shell: it snapshots the devices that already exist, waits for exactly one
//! NEW pending device to appear, approves it through the real add-confirm
//! device-lifecycle endpoint, and polls until that device is `active`.
//!
//! Every ambiguity fails CLOSED — zero candidates keep waiting until timeout,
//! more than one new pending device aborts without approving anything, a
//! terminal device state aborts, and a cancelled dev session stops promptly.
//! An aborted run only disables *auto*-approval; the local platform stays up so
//! the operator can still pair the device by hand.
//!
//! # Secret hygiene
//!
//! The bearer token and any `Authorization` value never appear in a log line,
//! an error, or the redacted device id. Device ids are truncated by
//! [`redact_device_id`] before they are logged.

use std::collections::HashSet;
use std::thread;
use std::time::{Duration, Instant};

use anyhow::{anyhow, bail, Context, Result};
use chrono::{DateTime, Utc};
use serde_json::Value;

use super::agent::client::AgentHttpClient;
use super::agent::session::AgentSession;
use crate::tui::{self, LogTx};

/// Total wall-clock budget for a single approval attempt (snapshot → confirm →
/// active). Shared across the wait-for-candidate and wait-for-active phases.
const APPROVAL_TIMEOUT: Duration = Duration::from_secs(120);
/// Delay between device-list polls.
const POLL_INTERVAL: Duration = Duration::from_millis(500);

/// A device registered to the owner's DID, in the minimal shape the
/// coordinator reasons about.
#[derive(Clone, Debug, PartialEq, Eq)]
struct ClientDevice {
    device_id: String,
    public_key: String,
    status: String,
    created_at: DateTime<Utc>,
}

/// Parse one device object from a `GET /api/v2/did-devices` entry.
///
/// `device_id`, `public_key_multibase`, `status`, and `created_at` are all
/// REQUIRED; a missing or malformed field is a hard error (fail closed).
/// `created_at` must be a valid RFC3339 timestamp. The real onboarding app
/// returns a CONSTANT platform `device_id` shared by every PWA registration, so
/// `public_key_multibase` is the only per-registration-unique identifier and is
/// what the coordinator keys candidate selection on. Error messages never echo a
/// bearer token.
fn parse_device(value: &Value) -> Result<ClientDevice> {
    let device_id = value
        .get("device_id")
        .and_then(Value::as_str)
        .ok_or_else(|| anyhow!("device payload missing 'device_id' string field"))?
        .to_string();
    let public_key = value
        .get("public_key_multibase")
        .and_then(Value::as_str)
        .ok_or_else(|| anyhow!("device payload missing 'public_key_multibase' string field"))?
        .to_string();
    let status = value
        .get("status")
        .and_then(Value::as_str)
        .ok_or_else(|| anyhow!("device payload missing 'status' string field"))?
        .to_string();
    let created_at_raw = value
        .get("created_at")
        .and_then(Value::as_str)
        .ok_or_else(|| anyhow!("device payload missing 'created_at' timestamp field"))?;
    let created_at = DateTime::parse_from_rfc3339(created_at_raw)
        .map_err(|error| anyhow!("device payload has malformed 'created_at' timestamp: {error}"))?
        .with_timezone(&Utc);
    Ok(ClientDevice {
        device_id,
        public_key,
        status,
        created_at,
    })
}

/// Select the single new pending device to approve, or fail closed.
///
/// A device is an eligible candidate iff it is `pending`, its
/// `public_key_multibase` was NOT present in the initial snapshot, and it was
/// created at or after activation. The public key — not the shared platform
/// `device_id` — is the per-registration-unique identity, so exclusion is keyed
/// on it. Exactly one eligible candidate returns `Ok(Some(..))`; none returns
/// `Ok(None)` (keep waiting); more than one (distinct new public keys) is an
/// error that refuses to auto-approve. The error contains no secret.
fn select_candidate(
    initial_keys: &HashSet<String>,
    activated_at: DateTime<Utc>,
    devices: &[ClientDevice],
) -> Result<Option<ClientDevice>> {
    let mut eligible = devices.iter().filter(|device| {
        device.status == "pending"
            && !initial_keys.contains(&device.public_key)
            && device.created_at >= activated_at
    });

    let Some(first) = eligible.next().cloned() else {
        return Ok(None);
    };
    if eligible.next().is_some() {
        bail!("multiple new pending devices appeared; refusing to auto-approve any of them");
    }
    Ok(Some(first))
}

/// Validate that `base_url` is a bare loopback HTTP origin.
///
/// Accepts ONLY `http://127.0.0.1:<port>` and `http://localhost:<port>` with a
/// numeric port. Rejects HTTPS, a missing port, user info (`user@`), a query
/// (`?`), a fragment (`#`), a path, and any non-loopback host. Plain string
/// parsing only — no URL-crate dependency.
fn validate_loopback_base_url(base_url: &str) -> Result<()> {
    let authority = base_url
        .strip_prefix("http://")
        .ok_or_else(|| anyhow!("operation mode requires an http:// loopback base URL"))?;

    if authority.contains('@') {
        bail!("operation-mode base URL must not contain user info");
    }
    if authority.contains('?') {
        bail!("operation-mode base URL must not contain a query string");
    }
    if authority.contains('#') {
        bail!("operation-mode base URL must not contain a fragment");
    }
    if authority.contains('/') {
        bail!("operation-mode base URL must not contain a path");
    }

    let (host, port) = authority
        .rsplit_once(':')
        .ok_or_else(|| anyhow!("operation-mode base URL must include an explicit port"))?;

    if host != "127.0.0.1" && host != "localhost" {
        bail!("operation-mode base URL host must be loopback (127.0.0.1 or localhost)");
    }
    if port.is_empty() || !port.bytes().all(|b| b.is_ascii_digit()) {
        bail!("operation-mode base URL must include a numeric port");
    }
    Ok(())
}

/// Redact a device id for logging: at most the first eight Unicode scalar
/// values, followed by an ellipsis. Never returns the full id.
fn redact_device_id(device_id: &str) -> String {
    let prefix: String = device_id.chars().take(8).collect();
    format!("{prefix}…")
}

/// The device-lifecycle surface the polling loop depends on. Implemented by the
/// real [`AgentHttpClient`] and, in tests, by a deterministic fake — so the
/// state machine can be driven without a live daemon.
trait DeviceApprovalApi {
    fn list(&self, token: &str) -> Result<Vec<Value>>;
    fn confirm(&self, token: &str, device_id: &str) -> Result<Value>;
}

impl DeviceApprovalApi for AgentHttpClient {
    fn list(&self, token: &str) -> Result<Vec<Value>> {
        self.list_did_devices(token)
    }

    fn confirm(&self, token: &str, device_id: &str) -> Result<Value> {
        self.confirm_did_device(token, device_id)
    }
}

/// Fetch and parse the current device list. A malformed entry fails the whole
/// poll (fail closed).
fn list_devices(api: &impl DeviceApprovalApi, token: &str) -> Result<Vec<ClientDevice>> {
    api.list(token)?
        .iter()
        .map(parse_device)
        .collect::<Result<Vec<_>>>()
}

/// The fail-closed approval state machine, parameterised for testing.
///
/// ```text
/// waiting -> exactly one new pending -> confirm -> wait for same id active -> success
/// waiting -> timeout                 -> error
/// waiting -> multiple new pending    -> error
/// active-wait -> rejected/revoked/expired -> error
/// any phase -> cancelled             -> error
/// ```
///
/// `cancel` is injected so the real run passes the process-global
/// `super::is_cancelled` while tests can force cancellation. Neither the token
/// nor any authorization value is ever placed in an error or a log line.
#[allow(clippy::too_many_arguments)] // one deterministic state machine; each knob is load-bearing and test-driven
fn poll_for_approval(
    api: &impl DeviceApprovalApi,
    token: &str,
    initial_keys: &HashSet<String>,
    activated_at: DateTime<Utc>,
    instance: &str,
    timeout: Duration,
    poll_interval: Duration,
    cancel: &dyn Fn() -> bool,
    log_tx: Option<&LogTx>,
) -> Result<String> {
    let deadline = Instant::now() + timeout;

    // Phase 1 — wait for exactly one new pending candidate, then confirm it.
    // The candidate is selected by its unique public key, but confirmed and
    // tracked by `device_id`: the server resolves the pending ADD operation from
    // the (constant, shared) platform device_id and flips the NEWEST matching
    // pending registration to active (proven live).
    let (device_id, public_key) = loop {
        if cancel() {
            bail!("operation-mode approval cancelled before a device was approved");
        }
        if Instant::now() >= deadline {
            bail!(
                "operation-mode approval timed out after {timeout:?} waiting for a new browser \
                 device for instance {instance}"
            );
        }

        let devices = list_devices(api, token)?;
        if let Some(candidate) = select_candidate(initial_keys, activated_at, &devices)? {
            break (candidate.device_id, candidate.public_key);
        }

        if cancel() {
            bail!("operation-mode approval cancelled before a device was approved");
        }
        thread::sleep(poll_interval);
    };

    // Log the redacted PUBLIC KEY (the genuinely-identifying value) immediately
    // before confirming — never the token, never the shared constant device_id.
    tui::sys_log(
        log_tx,
        format!(
            "client-node operation mode [{instance}]: approving device {}",
            redact_device_id(&public_key)
        ),
    );
    api.confirm(token, &device_id).with_context(|| {
        format!(
            "failed to confirm device {} for instance {instance}",
            redact_device_id(&public_key)
        )
    })?;

    // Phase 2 — wait for that same registration to become active, matched by its
    // UNIQUE public key (the shared constant device_id cannot distinguish it from
    // other pending registrations). Confirmation is idempotent: an already-active
    // device is success and is never re-confirmed. The returned identity is the
    // public key — the only per-registration-unique, and thus meaningful, value.
    loop {
        if cancel() {
            bail!("operation-mode approval cancelled while awaiting device activation");
        }
        if Instant::now() >= deadline {
            bail!(
                "operation-mode approval timed out after {timeout:?} waiting for device {} to \
                 become active for instance {instance}",
                redact_device_id(&public_key)
            );
        }

        let devices = list_devices(api, token)?;
        if let Some(device) = devices.into_iter().find(|d| d.public_key == public_key) {
            match device.status.as_str() {
                "active" => return Ok(public_key),
                "rejected" | "revoked" | "expired" => {
                    bail!(
                        "device {} entered terminal state '{}'; aborting operation-mode approval \
                         for instance {instance}",
                        redact_device_id(&public_key),
                        device.status
                    );
                }
                _ => {}
            }
        }

        if cancel() {
            bail!("operation-mode approval cancelled while awaiting device activation");
        }
        thread::sleep(poll_interval);
    }
}

/// How long the initial snapshot waits for the device surface to come up.
///
/// The snapshot runs as soon as the daemon answers on HTTPS, which is EARLIER
/// than the point at which every app has registered its capabilities — and
/// `GET /api/v2/did-devices` delegates to `onboarding.device.list`, so during
/// that window it answers 503. Treating that as fatal aborted the whole dev
/// session (daemon included) over a routine boot race; the surface typically
/// arrives a second or two later.
const SNAPSHOT_READY_TIMEOUT: Duration = Duration::from_secs(90);
/// Delay between snapshot readiness attempts.
const SNAPSHOT_RETRY_INTERVAL: Duration = Duration::from_millis(750);

/// `GET /api/v2/did-devices`, retried until the device surface is actually
/// serving. Only the *unavailable* shape is retried — anything else (a bad
/// token, a malformed body) is a real failure and is returned immediately, so
/// this cannot mask a genuine misconfiguration behind a long silent wait.
fn list_devices_once_ready(
    client: &AgentHttpClient,
    token: &str,
    timeout: Duration,
) -> Result<Vec<Value>> {
    let deadline = Instant::now() + timeout;
    loop {
        match client.list_did_devices(token) {
            Ok(devices) => return Ok(devices),
            Err(error) => {
                if !is_surface_unavailable(&error) {
                    return Err(error);
                }
                if Instant::now() >= deadline {
                    return Err(error.context(format!(
                        "device surface did not become available within {timeout:?}; \
                         onboarding.device.list never registered a provider"
                    )));
                }
                thread::sleep(SNAPSHOT_RETRY_INTERVAL);
            }
        }
    }
}

/// Does this error mean "not up yet" rather than "wrong"? The device route
/// answers 503 while `onboarding.*` has no provider, and the socket is simply
/// refused in the moments before the listener binds.
fn is_surface_unavailable(error: &anyhow::Error) -> bool {
    let text = format!("{error:#}");
    text.contains("503")
        || text.contains("Service Unavailable")
        || text.contains("SERVICE_UNAVAILABLE")
        || text.contains("Connection refused")
        || text.contains("connection refused")
}

/// A validated, snapshotted operation-mode run ready to spawn.
///
/// Built by [`prepare`]; consumed by [`PreparedOperationMode::spawn`], which
/// `dev::run`, `platform::run`, and `harness::probes::up` each call.
pub struct PreparedOperationMode {
    activated_at: DateTime<Utc>,
    client: AgentHttpClient,
    initial_keys: HashSet<String>,
    instance: String,
    session: AgentSession,
}

/// Validate the session's loopback base URL, then snapshot the public keys of
/// the DID's currently pending devices so a later poll can tell a genuinely NEW
/// registration apart from one that was already awaiting approval. The public
/// key — not the shared platform `device_id` — is snapshotted, because the real
/// onboarding app hands every PWA registration the same constant `device_id`.
///
/// `prepare(session)?.spawn(..)` is the entry point, wired into `dev::run`,
/// `platform::run`, and `harness::probes::up` for `node-app dev --client-node
/// --operation-mode` and `node-app harness up --client-node --operation-mode`.
pub fn prepare(session: AgentSession) -> Result<PreparedOperationMode> {
    validate_loopback_base_url(&session.base_url)?;
    let client = AgentHttpClient::new(session.base_url.clone());
    let initial = list_devices_once_ready(&client, &session.token, SNAPSHOT_READY_TIMEOUT)?;
    let initial_keys = initial
        .iter()
        .map(parse_device)
        .collect::<Result<Vec<_>>>()?
        .into_iter()
        .filter(|device| device.status == "pending")
        .map(|device| device.public_key)
        .collect();
    Ok(PreparedOperationMode {
        activated_at: Utc::now(),
        client,
        initial_keys,
        instance: session.instance.clone(),
        session,
    })
}

impl PreparedOperationMode {
    /// Run the real approval state machine against the live daemon, using the
    /// process-global cancellation signal. `log_tx` is threaded through to
    /// [`poll_for_approval`] so the "approving device …" line lands on the
    /// same TUI feedback channel as the "waiting" and "active" lines emitted
    /// around this call in [`spawn`].
    fn run(
        &self,
        log_tx: Option<&LogTx>,
        timeout: Duration,
        poll_interval: Duration,
    ) -> Result<String> {
        poll_for_approval(
            &self.client,
            &self.session.token,
            &self.initial_keys,
            self.activated_at,
            &self.instance,
            timeout,
            poll_interval,
            &super::is_cancelled,
            log_tx,
        )
    }

    /// Spawn the coordinator on a background thread. A polling failure disables
    /// auto-approval for this run but leaves the local platform available for
    /// manual pairing.
    pub fn spawn(self, log_tx: Option<LogTx>) -> std::thread::JoinHandle<()> {
        std::thread::spawn(move || {
            tui::sys_log(
                log_tx.as_ref(),
                format!(
                    "client-node operation mode [{}]: waiting for a new browser device",
                    self.instance
                ),
            );
            match self.run(log_tx.as_ref(), APPROVAL_TIMEOUT, POLL_INTERVAL) {
                Ok(device_id) => tui::sys_log(
                    log_tx.as_ref(),
                    format!(
                        "client-node operation mode [{}]: device {} active; browser may enter operation mode",
                        self.instance,
                        redact_device_id(&device_id)
                    ),
                ),
                Err(error) => tui::sys_log(
                    log_tx.as_ref(),
                    format!(
                        "client-node operation mode [{}] stopped: {error:#}",
                        self.instance
                    ),
                ),
            }
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::VecDeque;
    use std::sync::Mutex;

    use serde_json::json;

    /// The boot race that used to kill the whole dev session: the snapshot runs
    /// the moment the daemon answers on HTTPS, which is before `onboarding.*`
    /// has registered a provider, and `GET /api/v2/did-devices` answers 503
    /// until it does. "Not up yet" must be waited out; "wrong" must not be.
    #[test]
    fn only_the_unavailable_shape_is_treated_as_not_up_yet() {
        for transient in [
            "GET /api/v2/did-devices failed: http://127.0.0.1:3301/api/v2/did-devices: status code 503",
            "Service Unavailable",
            "{\"error\":{\"code\":\"SERVICE_UNAVAILABLE\"}}",
            "tcp connect error: Connection refused (os error 61)",
        ] {
            assert!(
                is_surface_unavailable(&anyhow!(transient.to_string())),
                "{transient:?} should be retried"
            );
        }

        for fatal in [
            "GET /api/v2/did-devices failed: status code 401",
            "Missing or invalid Authorization header",
            "device list response missing 'devices' array",
            "status code 500",
        ] {
            assert!(
                !is_surface_unavailable(&anyhow!(fatal.to_string())),
                "{fatal:?} must fail immediately, never be waited out"
            );
        }
    }

    fn device(id: &str, public_key: &str, status: &str, created_at: &str) -> ClientDevice {
        ClientDevice {
            device_id: id.into(),
            public_key: public_key.into(),
            status: status.into(),
            created_at: DateTime::parse_from_rfc3339(created_at)
                .unwrap()
                .with_timezone(&Utc),
        }
    }

    // ── select_candidate ─────────────────────────────────────────────────────

    #[test]
    fn selects_new_public_key_sharing_constant_device_id() {
        // Live-proven real-world case: the onboarding app returns a CONSTANT
        // platform device_id ("client_node_on_macos") shared by EVERY PWA
        // registration. A stale pending device (public key zStale) is in the
        // snapshot; a NEW browser registers with the SAME constant device_id but
        // a fresh unique public key (zFresh). It MUST be selected — the snapshot
        // is keyed on public key, so a shared device_id no longer masks a genuine
        // new registration. The selected device still carries the device_id used
        // to confirm.
        let initial = HashSet::from(["zStale".to_string()]);
        let selected = select_candidate(
            &initial,
            "2026-07-25T00:00:00Z".parse().unwrap(),
            &[device(
                "client_node_on_macos",
                "zFresh",
                "pending",
                "2026-07-25T08:38:33Z",
            )],
        )
        .unwrap()
        .unwrap();
        assert_eq!(selected.public_key, "zFresh");
        assert_eq!(selected.device_id, "client_node_on_macos");
    }

    #[test]
    fn ignores_candidates_present_in_initial_snapshot() {
        // A device whose PUBLIC KEY is in the snapshot is excluded even if its
        // device_id differs from anything snapshotted.
        let initial = HashSet::from(["zStaleKey".to_string()]);
        let selected = select_candidate(
            &initial,
            "2026-07-25T00:00:00Z".parse().unwrap(),
            &[device(
                "some_other_device_id",
                "zStaleKey",
                "pending",
                "2026-07-25T00:00:01Z",
            )],
        )
        .unwrap();
        assert_eq!(selected, None);
    }

    #[test]
    fn ignores_candidates_created_before_activation() {
        let selected = select_candidate(
            &HashSet::new(),
            "2026-07-25T00:00:10Z".parse().unwrap(),
            &[device("old", "zOld", "pending", "2026-07-25T00:00:09Z")],
        )
        .unwrap();
        assert_eq!(selected, None);
    }

    #[test]
    fn selects_exactly_one_new_pending_candidate() {
        let selected = select_candidate(
            &HashSet::new(),
            "2026-07-25T00:00:00Z".parse().unwrap(),
            &[device("fresh", "zFresh", "pending", "2026-07-25T00:00:01Z")],
        )
        .unwrap()
        .unwrap();
        assert_eq!(selected.public_key, "zFresh");
    }

    #[test]
    fn rejects_multiple_new_pending_candidates() {
        // Two distinct new public keys — even were they to share a device_id —
        // fail closed.
        let error = select_candidate(
            &HashSet::new(),
            "2026-07-25T00:00:00Z".parse().unwrap(),
            &[
                device(
                    "client_node_on_macos",
                    "zOne",
                    "pending",
                    "2026-07-25T00:00:01Z",
                ),
                device(
                    "client_node_on_macos",
                    "zTwo",
                    "pending",
                    "2026-07-25T00:00:02Z",
                ),
            ],
        )
        .unwrap_err();
        assert!(error.to_string().contains("multiple"));
        assert!(!error.to_string().contains("Bearer"));
    }

    #[test]
    fn accepts_candidate_created_at_the_activation_instant() {
        // `>=` boundary: a device created in the same instant as activation is a
        // candidate; the snapshot exclusion is the real staleness guard.
        let selected = select_candidate(
            &HashSet::new(),
            "2026-07-25T00:00:00Z".parse().unwrap(),
            &[device("edge", "zEdge", "pending", "2026-07-25T00:00:00Z")],
        )
        .unwrap()
        .unwrap();
        assert_eq!(selected.public_key, "zEdge");
    }

    #[test]
    fn ignores_active_devices_when_selecting() {
        // Only `pending` devices are candidates; a fresh, unsnapshotted `active`
        // device is not selected.
        let selected = select_candidate(
            &HashSet::new(),
            "2026-07-25T00:00:00Z".parse().unwrap(),
            &[device(
                "already",
                "zActive",
                "active",
                "2026-07-25T00:00:01Z",
            )],
        )
        .unwrap();
        assert_eq!(selected, None);
    }

    #[test]
    fn selects_device_sharing_a_snapshot_device_id_but_with_new_public_key() {
        // Vice-versa of the snapshot-exclusion test: a device that shares a
        // device_id with a snapshotted entry but carries a brand-new public key
        // IS selected. This is exactly the live failure mode — the constant
        // platform device_id must never mask a genuinely new registration.
        let initial = HashSet::from(["zSnapshotKey".to_string()]);
        let selected = select_candidate(
            &initial,
            "2026-07-25T00:00:00Z".parse().unwrap(),
            &[
                // Snapshotted registration, still pending (its key is excluded).
                device(
                    "client_node_on_macos",
                    "zSnapshotKey",
                    "pending",
                    "2026-07-25T05:25:24Z",
                ),
                // New registration, SAME device_id, fresh key → selected.
                device(
                    "client_node_on_macos",
                    "zNewKey",
                    "pending",
                    "2026-07-25T08:38:33Z",
                ),
            ],
        )
        .unwrap()
        .unwrap();
        assert_eq!(selected.public_key, "zNewKey");
        assert_eq!(selected.device_id, "client_node_on_macos");
    }

    // ── parse_device ─────────────────────────────────────────────────────────

    #[test]
    fn parse_device_reads_all_required_fields() {
        let parsed = parse_device(&json!({
            "device_id": "browser-abc",
            "public_key_multibase": "zBrowserAbcKey",
            "status": "pending",
            "created_at": "2026-07-25T00:00:01Z"
        }))
        .unwrap();
        assert_eq!(parsed.device_id, "browser-abc");
        assert_eq!(parsed.public_key, "zBrowserAbcKey");
        assert_eq!(parsed.status, "pending");
    }

    #[test]
    fn parse_device_rejects_missing_created_at() {
        let error = parse_device(&json!({
            "device_id": "browser-abc",
            "public_key_multibase": "zBrowserAbcKey",
            "status": "pending"
        }))
        .unwrap_err();
        assert!(error.to_string().contains("created_at"));
    }

    #[test]
    fn parse_device_rejects_malformed_created_at() {
        let error = parse_device(&json!({
            "device_id": "browser-abc",
            "public_key_multibase": "zBrowserAbcKey",
            "status": "pending",
            "created_at": "not-a-timestamp"
        }))
        .unwrap_err();
        assert!(error.to_string().contains("created_at"));
    }

    #[test]
    fn parse_device_rejects_missing_device_id() {
        let error = parse_device(&json!({
            "public_key_multibase": "zBrowserAbcKey",
            "status": "pending",
            "created_at": "2026-07-25T00:00:01Z"
        }))
        .unwrap_err();
        assert!(error.to_string().contains("device_id"));
    }

    #[test]
    fn parse_device_rejects_missing_public_key_multibase() {
        let error = parse_device(&json!({
            "device_id": "browser-abc",
            "status": "pending",
            "created_at": "2026-07-25T00:00:01Z"
        }))
        .unwrap_err();
        assert!(error.to_string().contains("public_key_multibase"));
    }

    // ── validate_loopback_base_url ───────────────────────────────────────────

    #[test]
    fn accepts_loopback_ipv4_with_port() {
        validate_loopback_base_url("http://127.0.0.1:3001").unwrap();
    }

    #[test]
    fn accepts_localhost_with_port() {
        validate_loopback_base_url("http://localhost:8080").unwrap();
    }

    #[test]
    fn rejects_https_scheme() {
        assert!(validate_loopback_base_url("https://127.0.0.1:3001").is_err());
    }

    #[test]
    fn rejects_missing_port() {
        assert!(validate_loopback_base_url("http://127.0.0.1").is_err());
        assert!(validate_loopback_base_url("http://localhost").is_err());
    }

    #[test]
    fn rejects_empty_port() {
        assert!(validate_loopback_base_url("http://127.0.0.1:").is_err());
    }

    #[test]
    fn rejects_non_numeric_port() {
        assert!(validate_loopback_base_url("http://127.0.0.1:abc").is_err());
    }

    #[test]
    fn rejects_user_info() {
        assert!(validate_loopback_base_url("http://user@127.0.0.1:3001").is_err());
    }

    #[test]
    fn rejects_query_string() {
        assert!(validate_loopback_base_url("http://127.0.0.1:3001?token=x").is_err());
    }

    #[test]
    fn rejects_fragment() {
        assert!(validate_loopback_base_url("http://127.0.0.1:3001#frag").is_err());
    }

    #[test]
    fn rejects_path() {
        assert!(validate_loopback_base_url("http://127.0.0.1:3001/api").is_err());
    }

    #[test]
    fn rejects_non_loopback_host() {
        assert!(validate_loopback_base_url("http://evil.example.com:3001").is_err());
        assert!(validate_loopback_base_url("http://10.0.0.5:3001").is_err());
    }

    // ── redact_device_id ─────────────────────────────────────────────────────

    #[test]
    fn redacts_long_device_id_to_eight_chars_plus_ellipsis() {
        let redacted = redact_device_id("browser-device-1234567890");
        assert_eq!(redacted, "browser-…");
        assert!(!redacted.contains("device-1234567890"));
    }

    #[test]
    fn redacts_short_device_id_without_panicking_on_unicode() {
        // Fewer than eight scalar values plus multibyte input must not panic on
        // a byte boundary.
        let redacted = redact_device_id("dév");
        assert!(redacted.starts_with("dév"));
        assert!(redacted.ends_with('…'));
    }

    // ── poll_for_approval state machine (deterministic fake) ─────────────────

    /// One scripted list outcome.
    enum ListStep {
        Ok(Vec<Value>),
        Err(String),
    }

    /// Deterministic [`DeviceApprovalApi`] fake: a queue of scripted list
    /// outcomes (the last one sticks once exhausted) and a recorded log of
    /// confirmations. `confirm_err` forces `confirm` to fail.
    struct FakeApi {
        steps: Mutex<VecDeque<ListStep>>,
        last: Mutex<Vec<Value>>,
        confirmations: Mutex<Vec<String>>,
        confirm_err: Option<String>,
    }

    impl FakeApi {
        fn new(steps: Vec<ListStep>) -> Self {
            Self {
                steps: Mutex::new(steps.into_iter().collect()),
                last: Mutex::new(Vec::new()),
                confirmations: Mutex::new(Vec::new()),
                confirm_err: None,
            }
        }

        fn with_confirm_error(mut self, message: &str) -> Self {
            self.confirm_err = Some(message.to_string());
            self
        }

        fn confirmations(&self) -> Vec<String> {
            self.confirmations.lock().unwrap().clone()
        }
    }

    impl DeviceApprovalApi for FakeApi {
        fn list(&self, _token: &str) -> Result<Vec<Value>> {
            let mut steps = self.steps.lock().unwrap();
            match steps.pop_front() {
                Some(ListStep::Ok(devices)) => {
                    *self.last.lock().unwrap() = devices.clone();
                    Ok(devices)
                }
                Some(ListStep::Err(message)) => Err(anyhow!(message)),
                // Queue exhausted: keep returning the last observed list so the
                // active-wait phase can settle deterministically.
                None => Ok(self.last.lock().unwrap().clone()),
            }
        }

        fn confirm(&self, _token: &str, device_id: &str) -> Result<Value> {
            self.confirmations
                .lock()
                .unwrap()
                .push(device_id.to_string());
            match &self.confirm_err {
                Some(message) => Err(anyhow!(message.clone())),
                None => Ok(json!({ "device_id": device_id, "status": "confirming" })),
            }
        }
    }

    fn dev_json(id: &str, public_key: &str, status: &str, created_at: &str) -> Value {
        json!({
            "device_id": id,
            "public_key_multibase": public_key,
            "status": status,
            "created_at": created_at
        })
    }

    fn no_cancel() -> bool {
        false
    }

    const ACTIVATED_AT: &str = "2026-07-25T00:00:00Z";
    /// The constant platform device_id the real onboarding app hands every PWA
    /// registration — shared across all pending devices; only the public key
    /// distinguishes them.
    const PLATFORM_DEVICE_ID: &str = "client_node_on_macos";

    fn run_poll(
        api: &impl DeviceApprovalApi,
        cancel: &dyn Fn() -> bool,
        timeout: Duration,
    ) -> Result<String> {
        poll_for_approval(
            api,
            "secret-bearer-token-value",
            &HashSet::new(),
            ACTIVATED_AT.parse().unwrap(),
            "alice",
            timeout,
            Duration::from_millis(1),
            cancel,
            None,
        )
    }

    #[test]
    fn zero_candidates_times_out_without_confirming() {
        let api = FakeApi::new(vec![ListStep::Ok(vec![])]);
        let error = run_poll(&api, &no_cancel, Duration::from_millis(20)).unwrap_err();
        assert!(error.to_string().contains("timed out"));
        assert!(api.confirmations().is_empty());
    }

    #[test]
    fn confirms_single_candidate_exactly_once_then_succeeds() {
        // The device_id is the shared constant; the public key is what tracks
        // and identifies the registration. Confirm targets the device_id.
        let api = FakeApi::new(vec![
            ListStep::Ok(vec![dev_json(
                PLATFORM_DEVICE_ID,
                "zBrowserX",
                "pending",
                "2026-07-25T00:00:01Z",
            )]),
            ListStep::Ok(vec![dev_json(
                PLATFORM_DEVICE_ID,
                "zBrowserX",
                "confirming",
                "2026-07-25T00:00:01Z",
            )]),
            ListStep::Ok(vec![dev_json(
                PLATFORM_DEVICE_ID,
                "zBrowserX",
                "active",
                "2026-07-25T00:00:01Z",
            )]),
        ]);
        let public_key = run_poll(&api, &no_cancel, Duration::from_secs(5)).unwrap();
        assert_eq!(public_key, "zBrowserX");
        // Confirmed by the (constant) device_id — proven-correct target.
        assert_eq!(api.confirmations(), vec![PLATFORM_DEVICE_ID.to_string()]);
    }

    #[test]
    fn candidate_already_active_after_confirmation_succeeds() {
        let api = FakeApi::new(vec![
            ListStep::Ok(vec![dev_json(
                PLATFORM_DEVICE_ID,
                "zBrowserX",
                "pending",
                "2026-07-25T00:00:01Z",
            )]),
            ListStep::Ok(vec![dev_json(
                PLATFORM_DEVICE_ID,
                "zBrowserX",
                "active",
                "2026-07-25T00:00:01Z",
            )]),
        ]);
        let public_key = run_poll(&api, &no_cancel, Duration::from_secs(5)).unwrap();
        assert_eq!(public_key, "zBrowserX");
        // Confirmed once — never re-confirmed after it turned active.
        assert_eq!(api.confirmations(), vec![PLATFORM_DEVICE_ID.to_string()]);
    }

    #[test]
    fn multiple_candidates_abort_without_confirming() {
        // Two new registrations sharing the constant device_id but with distinct
        // public keys — fail closed on the two distinct keys.
        let api = FakeApi::new(vec![ListStep::Ok(vec![
            dev_json(
                PLATFORM_DEVICE_ID,
                "zBrowserA",
                "pending",
                "2026-07-25T00:00:01Z",
            ),
            dev_json(
                PLATFORM_DEVICE_ID,
                "zBrowserB",
                "pending",
                "2026-07-25T00:00:02Z",
            ),
        ])]);
        let error = run_poll(&api, &no_cancel, Duration::from_secs(5)).unwrap_err();
        assert!(error.to_string().contains("multiple"));
        assert!(api.confirmations().is_empty());
    }

    #[test]
    fn terminal_state_after_confirmation_aborts() {
        for terminal in ["rejected", "revoked", "expired"] {
            let api = FakeApi::new(vec![
                ListStep::Ok(vec![dev_json(
                    PLATFORM_DEVICE_ID,
                    "zBrowserX",
                    "pending",
                    "2026-07-25T00:00:01Z",
                )]),
                ListStep::Ok(vec![dev_json(
                    PLATFORM_DEVICE_ID,
                    "zBrowserX",
                    terminal,
                    "2026-07-25T00:00:01Z",
                )]),
            ]);
            let error = run_poll(&api, &no_cancel, Duration::from_secs(5)).unwrap_err();
            assert!(
                error.to_string().contains(terminal),
                "expected terminal state '{terminal}' in error: {error}"
            );
            assert_eq!(api.confirmations(), vec![PLATFORM_DEVICE_ID.to_string()]);
        }
    }

    #[test]
    fn cancellation_stops_before_confirming() {
        let api = FakeApi::new(vec![ListStep::Ok(vec![dev_json(
            PLATFORM_DEVICE_ID,
            "zBrowserX",
            "pending",
            "2026-07-25T00:00:01Z",
        )])]);
        let error = run_poll(&api, &|| true, Duration::from_secs(5)).unwrap_err();
        assert!(error.to_string().contains("cancelled"));
        assert!(api.confirmations().is_empty());
    }

    #[test]
    fn malformed_response_fails_closed() {
        let api = FakeApi::new(vec![ListStep::Ok(vec![json!({
            "device_id": PLATFORM_DEVICE_ID,
            "public_key_multibase": "zBrowserX",
            "status": "pending"
            // no created_at
        })])]);
        let error = run_poll(&api, &no_cancel, Duration::from_secs(5)).unwrap_err();
        assert!(error.to_string().contains("created_at"));
        assert!(api.confirmations().is_empty());
    }

    #[test]
    fn missing_public_key_fails_closed() {
        // A device list entry with no public_key_multibase fails the whole poll.
        let api = FakeApi::new(vec![ListStep::Ok(vec![json!({
            "device_id": PLATFORM_DEVICE_ID,
            "status": "pending",
            "created_at": "2026-07-25T00:00:01Z"
            // no public_key_multibase
        })])]);
        let error = run_poll(&api, &no_cancel, Duration::from_secs(5)).unwrap_err();
        assert!(error.to_string().contains("public_key_multibase"));
        assert!(api.confirmations().is_empty());
    }

    #[test]
    fn list_transport_error_fails_closed() {
        let api = FakeApi::new(vec![ListStep::Err("network down".to_string())]);
        let error = run_poll(&api, &no_cancel, Duration::from_secs(5)).unwrap_err();
        assert!(error.to_string().contains("network down"));
        assert!(api.confirmations().is_empty());
    }

    #[test]
    fn confirmation_error_does_not_disclose_token() {
        let api = FakeApi::new(vec![ListStep::Ok(vec![dev_json(
            PLATFORM_DEVICE_ID,
            "zBrowserX",
            "pending",
            "2026-07-25T00:00:01Z",
        )])])
        .with_confirm_error("simulated confirm failure");
        let error = run_poll(&api, &no_cancel, Duration::from_secs(5)).unwrap_err();
        let text = format!("{error:#}");
        assert!(text.contains("simulated confirm failure"));
        // The bearer token must never surface in a propagated error.
        assert!(!text.contains("secret-bearer-token-value"));
        assert!(!text.contains("Bearer"));
        // The confirm was attempted exactly once against the device_id.
        assert_eq!(api.confirmations(), vec![PLATFORM_DEVICE_ID.to_string()]);
    }

    #[test]
    fn logs_redacted_public_key_before_confirming_without_leaking_secrets() {
        use crate::tui::TuiEvent;

        let (tx, rx) = std::sync::mpsc::sync_channel(16);
        // Constant device_id + a unique, sensitive public key. The redacted
        // audit line must carry the public key prefix (the meaningful id), not
        // the shared constant device_id, and never the full key or the token.
        let api = FakeApi::new(vec![
            ListStep::Ok(vec![dev_json(
                PLATFORM_DEVICE_ID,
                "browser-key-secret-9999",
                "pending",
                "2026-07-25T00:00:01Z",
            )]),
            ListStep::Ok(vec![dev_json(
                PLATFORM_DEVICE_ID,
                "browser-key-secret-9999",
                "active",
                "2026-07-25T00:00:01Z",
            )]),
        ]);
        let public_key = poll_for_approval(
            &api,
            "secret-bearer-token-value",
            &HashSet::new(),
            ACTIVATED_AT.parse().unwrap(),
            "alice",
            Duration::from_secs(5),
            Duration::from_millis(1),
            &no_cancel,
            Some(&tx),
        )
        .unwrap();
        assert_eq!(public_key, "browser-key-secret-9999");

        drop(tx);
        let lines: Vec<String> = rx
            .into_iter()
            .filter_map(|event| match event {
                TuiEvent::Log(entry) => Some(entry.line),
                _ => None,
            })
            .collect();
        let approving: Vec<&String> = lines.iter().filter(|l| l.contains("approving")).collect();
        assert_eq!(
            approving.len(),
            1,
            "expected exactly one approving log line, got {lines:?}"
        );
        let line = approving[0];
        assert!(
            line.contains("browser-…"),
            "approving line must carry the redacted public key: {line}"
        );
        // Neither the full public key nor the token may surface in a log line.
        assert!(!line.contains("secret-9999"));
        assert!(!line.contains("secret-bearer-token-value"));
    }

    /// Proves the PRODUCTION path (what `PreparedOperationMode::run`, called
    /// from `spawn`, actually does with a real `Some(log_tx)`): the
    /// "approving device …" line (#2) lands on the SAME channel as the
    /// "waiting" (#1) / "active" (#3) lines that `spawn` logs directly, the
    /// id is redacted, and no token or `Bearer` value ever appears on that
    /// channel. This exercises `poll_for_approval` with the exact
    /// `log_tx`/timeout/poll_interval shape `run` forwards — the only piece
    /// `run` cannot itself be unit-tested with is constructing a live
    /// `AgentHttpClient`.
    #[test]
    fn production_path_delivers_all_three_lifecycle_lines_on_one_channel_without_leaking_secrets() {
        use crate::tui::TuiEvent;

        let (tx, rx) = std::sync::mpsc::sync_channel(16);
        // Constant shared device_id; the sensitive per-registration public key
        // is the value that gets redacted onto the channel.
        let api = FakeApi::new(vec![
            ListStep::Ok(vec![dev_json(
                PLATFORM_DEVICE_ID,
                "browser-key-secret-abcdef1234",
                "pending",
                "2026-07-25T00:00:01Z",
            )]),
            ListStep::Ok(vec![dev_json(
                PLATFORM_DEVICE_ID,
                "browser-key-secret-abcdef1234",
                "active",
                "2026-07-25T00:00:01Z",
            )]),
        ]);
        let instance = "alice";
        const TOKEN: &str = "secret-bearer-token-value";

        // Mirror exactly what `spawn` does around `run`: line #1 before, the
        // state machine (which now carries `log_tx` all the way through,
        // exactly as `run` does) in the middle, and line #3 after — all on
        // the one `tx`/`LogTx` channel.
        tui::sys_log(
            Some(&tx),
            format!("client-node operation mode [{instance}]: waiting for a new browser device"),
        );
        let device_id = poll_for_approval(
            &api,
            TOKEN,
            &HashSet::new(),
            ACTIVATED_AT.parse().unwrap(),
            instance,
            Duration::from_secs(5),
            Duration::from_millis(1),
            &no_cancel,
            Some(&tx),
        )
        .unwrap();
        tui::sys_log(
            Some(&tx),
            format!(
                "client-node operation mode [{instance}]: device {} active; browser may enter operation mode",
                redact_device_id(&device_id)
            ),
        );

        drop(tx);
        let lines: Vec<String> = rx
            .into_iter()
            .filter_map(|event| match event {
                TuiEvent::Log(entry) => Some(entry.line),
                _ => None,
            })
            .collect();

        assert_eq!(
            lines.len(),
            3,
            "expected all three lifecycle lines on one channel: {lines:?}"
        );
        assert!(lines[0].contains("waiting for a new browser device"));
        assert!(
            lines[1].contains("approving device") && lines[1].contains("browser-…"),
            "line #2 must be the redacted approving line: {}",
            lines[1]
        );
        assert!(
            lines[2].contains("active; browser may enter operation mode")
                && lines[2].contains("browser-…"),
            "line #3 must reference the same redacted id: {}",
            lines[2]
        );

        // No line on the shared channel ever carries the full device id, the
        // bearer token, or the literal "Bearer" scheme.
        for line in &lines {
            assert!(
                !line.contains("secret-abcdef1234"),
                "leaked full device id: {line}"
            );
            assert!(!line.contains(TOKEN), "leaked bearer token: {line}");
            assert!(!line.contains("Bearer"), "leaked Bearer scheme: {line}");
        }
    }

    /// Guards the wiring itself: `PreparedOperationMode::run` must accept a
    /// `log_tx` parameter and forward it into `poll_for_approval` — it must
    /// NOT hardcode `None`, and `spawn` must call it with `log_tx.as_ref()`.
    /// `run` can't be driven directly in a unit test without a live
    /// `AgentHttpClient` (see the seam-level test above for the behavioral
    /// proof), so this locks the exact call shapes at the source level to
    /// catch a regression back to the hardcoded-`None` bug.
    #[test]
    fn run_forwards_log_tx_into_poll_for_approval_instead_of_hardcoding_none() {
        let source = include_str!("operation_mode.rs");
        assert!(
            source.contains("fn run(") && source.contains("log_tx: Option<&LogTx>"),
            "PreparedOperationMode::run must accept a log_tx: Option<&LogTx> parameter"
        );
        assert!(
            source.contains("self.run(log_tx.as_ref(), APPROVAL_TIMEOUT, POLL_INTERVAL)"),
            "spawn must call run with log_tx.as_ref(), not a hardcoded None"
        );

        // The `run` method body (between its signature and the closing brace
        // of `impl PreparedOperationMode`'s next method) must pass the real
        // `log_tx` into `poll_for_approval`, never a literal `None`.
        let run_start = source.find("fn run(").expect("run method must exist");
        let run_end = source[run_start..]
            .find("/// Spawn the coordinator")
            .map(|offset| run_start + offset)
            .expect("spawn method must follow run");
        let run_body = &source[run_start..run_end];
        assert!(
            run_body.contains("poll_for_approval("),
            "run must call poll_for_approval: {run_body}"
        );
        assert!(
            run_body.contains("log_tx,"),
            "run must forward log_tx into poll_for_approval: {run_body}"
        );
        assert!(
            !run_body.contains("&super::is_cancelled,\n            None,"),
            "run must not hardcode log_tx: None when calling poll_for_approval: {run_body}"
        );
    }
}