node-app-build 7.1.10

//! Minimal blocking HTTP client wrapping the daemon's auth endpoints.
//!
//! The minimal core (#2939) serves first-time owner onboarding and nothing
//! else of the pre-cut auth surface:
//!   POST /api/auth/onboarding-challenge
//!   POST /api/auth/complete-onboarding
//!   GET  /.well-known/client-node-origin    (node_id, public)
//! `/api/auth/challenge`, `/api/auth/verify`, `/api/node/info` and
//! `/api/v2/internal/test/seed-peer` are gone; a saved session is reused and
//! peers are introduced by gossip card (`agent::cross_seed_peers`).
//!
//! All daemon responses are wrapped in an `ApiResponse { success, data,
//! error }` envelope on most routes, but the v2 hexagonal routes return
//! domain objects directly. `extract_data` accepts both shapes — same
//! tolerance the e2e `unwrapResponseOrThrow` helper uses.
//!
//! ## Token refresh (spec D5)
//!
//! Every authed call (GET or POST with a bearer token) goes through
//! [`AgentHttpClient::authed_get`] / [`authed_post`][AgentHttpClient::authed_post],
//! which centralize 401 handling: on a 401, and only when the client was
//! built via [`AgentHttpClient::with_session`], the bound `AgentSession` is
//! reloaded, its `refresh_token` is exchanged at `POST /api/auth/refresh` —
//! or, on the minimal core, which has no such route, the owner token is
//! re-issued from the instance's own `JWT_SECRET`
//! ([`reissue_from_instance_secret`]) — the new token is persisted back to
//! disk, and the original request is retried exactly once with it. `JWT_TOKEN_EXPIRY`
//! defaults to 900s (`core/foundation/src/config.rs`), so without this any
//! harness session running longer than 15 minutes would fail every
//! subsequent probe — this is what `harness status` was actually reporting
//! as a healthy `{"error": "Invalid or expired token"}` before spec D5.

use anyhow::{anyhow, bail, Context, Result};
use serde_json::{json, Value};
use std::path::PathBuf;
use std::sync::Mutex;

const AUTH_TIMEOUT_SECS: u64 = 30;

pub struct AgentHttpClient {
    base_url: String,
    agent: ureq::Agent,
    /// The on-disk `AgentSession` this client may refresh against. `None`
    /// for the handful of call sites that run before any session exists
    /// (onboarding itself) or that intentionally bypass refresh (the L402
    /// cross-node probe's raw-status path). Every other constructor should
    /// go through [`Self::with_session`].
    session_path: Option<PathBuf>,
    /// The freshest access token this client has seen, updated in place
    /// after a successful refresh. Consulted before the caller-supplied
    /// token on every authed call so that, within one probe that issues
    /// several requests (e.g. `harness status`'s balance/channels/peers),
    /// only the FIRST request pays the refresh round trip.
    current_token: Mutex<Option<String>>,
}

pub struct OnboardingChallenge {
    pub challenge_id: String,
    pub challenge: String,
}

pub struct AuthSession {
    pub token: String,
    pub refresh_token: String,
}

/// A classified failure from an authed request — distinct enough that a
/// caller (`harness status`, in particular) can tell "the token was
/// rejected" apart from "the daemon could not be reached at all" rather
/// than collapsing both into an opaque `{"error": "..."}` string (spec D5 /
/// AC-0: a probe must never report a health check as healthy, or as an
/// undifferentiated failure, when the actual failure mode is knowable).
///
/// The `Display` impl is deliberately self-contained (method, path, and
/// full body/detail in one flat string) rather than relying on `anyhow`
/// context layering, so `to_string()` on the resulting error never loses
/// detail that a caller (e.g. `harness pay`'s "no route" detection) greps
/// for in the message text.
#[derive(Debug)]
pub enum ProbeError {
    /// 401 from an authed endpoint. If this is returned to a caller at
    /// all, refreshing (via a client built with [`AgentHttpClient::with_session`])
    /// either was not possible (no bound session) or did not recover it
    /// (the refresh call itself failed, or the retried request 401'd
    /// again — meaning the refresh token itself is dead).
    Unauthorized {
        method: &'static str,
        path: String,
        detail: String,
    },
    /// A non-401, non-2xx response.
    Http {
        method: &'static str,
        path: String,
        status: u16,
        body: String,
    },
    /// The request could not reach the daemon at all (connection refused,
    /// DNS failure, timeout) — the daemon may simply not be running.
    Unreachable {
        method: &'static str,
        path: String,
        detail: String,
    },
    /// A 2xx response whose body did not parse as JSON.
    Protocol {
        method: &'static str,
        path: String,
        detail: String,
    },
}

impl std::fmt::Display for ProbeError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ProbeError::Unauthorized { method, path, detail } => {
                write!(f, "{method} {path} unauthorized: {detail}")
            }
            ProbeError::Http { method, path, status, body } => {
                write!(f, "{method} {path} returned HTTP {status}: {body}")
            }
            ProbeError::Unreachable { method, path, detail } => {
                write!(f, "{method} {path} unreachable: {detail}")
            }
            ProbeError::Protocol { method, path, detail } => {
                write!(f, "{method} {path}: could not parse response JSON: {detail}")
            }
        }
    }
}

impl std::error::Error for ProbeError {}

fn default_agent() -> ureq::Agent {
    ureq::AgentBuilder::new()
        .timeout(std::time::Duration::from_secs(AUTH_TIMEOUT_SECS))
        .build()
}

impl AgentHttpClient {
    pub fn new(base_url: impl Into<String>) -> Self {
        Self {
            base_url: base_url.into(),
            agent: default_agent(),
            session_path: None,
            current_token: Mutex::new(None),
        }
    }

    /// Like [`Self::new`], but binds the on-disk `AgentSession` at
    /// `session_path` so every authed call this client makes can recover a
    /// 401 by refreshing — see the module-level docs.
    pub fn with_session(base_url: impl Into<String>, session_path: PathBuf) -> Self {
        Self {
            base_url: base_url.into(),
            agent: default_agent(),
            session_path: Some(session_path),
            current_token: Mutex::new(None),
        }
    }

    /// Returns true iff the node has not yet been onboarded by anyone.
    /// Tolerates schema drift: any non-200 is treated as "unknown — try
    /// onboarding and let the server reject" so we don't lock the agent
    /// out on an unrelated upstream change.
    pub fn is_unowned(&self) -> Result<bool> {
        let response = self
            .agent
            .get(&format!("{}/api/auth/onboarding-status", self.base_url))
            .call();
        match response {
            Ok(r) => {
                let value: Value = r.into_json().context("parse onboarding-status JSON")?;
                let data = extract_data(value);
                // Field name is `is_onboarded` on the OnboardingStatusResponse
                // shape; missing == treat as fresh.
                let onboarded = data
                    .get("is_onboarded")
                    .and_then(Value::as_bool)
                    .unwrap_or(false);
                Ok(!onboarded)
            }
            Err(ureq::Error::Status(_, _)) => Ok(false),
            Err(e) => Err(anyhow!("onboarding-status request failed: {e}")),
        }
    }

    pub fn create_onboarding_challenge(&self) -> Result<OnboardingChallenge> {
        let value: Value = post_json(
            &self.agent,
            &format!("{}/api/auth/onboarding-challenge", self.base_url),
            None,
            &json!({}),
        )?;
        let data = extract_data(value);
        Ok(OnboardingChallenge {
            challenge_id: string_field(&data, "challenge_id")?,
            challenge: string_field(&data, "challenge")?,
        })
    }

    pub fn complete_onboarding(
        &self,
        public_key_hex: &str,
        challenge_id: &str,
        signature_hex: &str,
        username: &str,
    ) -> Result<AuthSession> {
        let value: Value = post_json(
            &self.agent,
            &format!("{}/api/auth/complete-onboarding", self.base_url),
            None,
            &json!({
                "public_key": public_key_hex,
                "challenge_id": challenge_id,
                "signature": signature_hex,
                "username": username,
                "first_name": "Agent",
                "last_name": "Dev",
            }),
        )?;
        let data = extract_data(value);
        Ok(AuthSession {
            token: string_field(&data, "token")?,
            refresh_token: string_field(&data, "refresh_token")?,
        })
    }

    /// POST /api/auth/refresh — exchange a refresh token for a fresh
    /// access/refresh token pair. Public so it is independently
    /// unit-testable; the 401-recovery path ([`Self::refresh_and_persist`])
    /// is the only internal caller today.
    pub fn refresh(&self, refresh_token: &str) -> Result<AuthSession> {
        let value: Value = post_json(
            &self.agent,
            &format!("{}/api/auth/refresh", self.base_url),
            None,
            &json!({ "refresh_token": refresh_token }),
        )?;
        let data = extract_data(value);
        Ok(AuthSession {
            token: string_field(&data, "token")?,
            refresh_token: string_field(&data, "refresh_token")?,
        })
    }

    /// This node's id — its Lightning public key, which the minimal core
    /// resolves from the ldk-node app and proves at the public
    /// `GET /.well-known/client-node-origin` (the pre-cut `/api/node/info`
    /// is gone, #2939). Empty, or an error, while ldk-node is still starting.
    pub fn get_node_id(&self) -> Result<String> {
        let path = "/.well-known/client-node-origin";
        let response = self
            .agent
            .get(&format!("{}{path}", self.base_url))
            .call();
        let value = Self::classify_response("GET", path, response).map_err(anyhow::Error::new)?;
        string_field(&extract_data(value), "node_id")
    }

    /// Prove `token` is accepted by an owner-gated route, refreshing it once
    /// on a 401 when this client is bound to a session file.
    ///
    /// `GET /api/v2/client-transport/candidates` is behind the owner gate on
    /// the minimal core and has no side effects — the cheapest authed read
    /// it serves.
    pub fn probe_token(&self, token: &str) -> Result<()> {
        self.authed_get("/api/v2/client-transport/candidates", token)
            .map(|_| ())
    }

    // ── Lightning / capability probe methods ─────────────────────────────────

    /// GET /api/balance — wallet balance summary.
    pub fn get_balance(&self, token: &str) -> Result<Value> {
        Ok(extract_data(self.authed_get("/api/balance", token)?))
    }

    /// GET /api/channels — list open Lightning channels.
    pub fn list_channels(&self, token: &str) -> Result<Value> {
        Ok(extract_data(self.authed_get("/api/channels", token)?))
    }

    /// GET /api/peers — list connected Lightning peers.
    pub fn list_peers(&self, token: &str) -> Result<Value> {
        Ok(extract_data(self.authed_get("/api/peers", token)?))
    }

    /// GET /api/onchain/address/new — generate a fresh on-chain Bitcoin address.
    ///
    /// Returns the `address` string from the response.
    pub fn new_onchain_address(&self, token: &str) -> Result<String> {
        let data = extract_data(self.authed_get("/api/onchain/address/new", token)?);
        string_field(&data, "address")
    }

    /// POST /api/peers/connect — connect to a Lightning peer.
    ///
    /// `peer` must be in `pubkey@host:port` format (the server parses it).
    pub fn connect_peer(&self, token: &str, peer: &str) -> Result<Value> {
        Ok(extract_data(self.authed_post(
            "/api/peers/connect",
            token,
            &json!({ "peer_info": peer }),
        )?))
    }

    /// Build the JSON body for `POST /api/channels/open`.
    ///
    /// Extracted so the body shape can be unit-tested without a live daemon.
    /// The `peer_pubkey_and_address` field carries the `pubkey@host:port`
    /// string and the server splits it internally.
    pub(crate) fn open_channel_body(peer: &str, sats: u64, push_msat: u64) -> Value {
        json!({
            "peer_pubkey_and_address": peer,
            "channel_amount_sats": sats,
            "push_to_counterparty_msat": push_msat,
        })
    }

    /// POST /api/channels/open — open a Lightning channel with a peer.
    ///
    /// `peer` must be in `pubkey@host:port` format.
    pub fn open_channel(
        &self,
        token: &str,
        peer: &str,
        sats: u64,
        push_msat: u64,
    ) -> Result<Value> {
        let body = Self::open_channel_body(peer, sats, push_msat);
        Ok(extract_data(self.authed_post("/api/channels/open", token, &body)?))
    }

    /// POST /api/payments/invoices/create — create a BOLT11 invoice.
    ///
    /// Returns the bolt11 string from the `invoice` field of the response.
    pub fn create_invoice(
        &self,
        token: &str,
        amount_msat: u64,
        memo: &str,
    ) -> Result<String> {
        let data = extract_data(self.authed_post(
            "/api/payments/invoices/create",
            token,
            &json!({
                "amount_msat": amount_msat,
                "description": memo,
                "expiry_secs": 3600u32,
            }),
        )?);
        string_field(&data, "invoice")
    }

    /// POST /api/payments/send/invoice — pay a BOLT11 invoice.
    ///
    /// The `invoice` field carries the bolt11 string.
    pub fn pay_invoice(&self, token: &str, bolt11: &str) -> Result<Value> {
        Ok(extract_data(self.authed_post(
            "/api/payments/send/invoice",
            token,
            &json!({ "invoice": bolt11 }),
        )?))
    }

    /// POST /api/v2/system/capabilities/invoke — invoke a registered capability.
    ///
    /// Routes through the platform capability router; works for both core and
    /// builtin-app capabilities.
    pub fn invoke_capability(
        &self,
        token: &str,
        capability: &str,
        payload: Value,
    ) -> Result<Value> {
        Ok(extract_data(self.authed_post(
            "/api/v2/system/capabilities/invoke",
            token,
            &json!({
                "capability": capability,
                "payload": payload,
            }),
        )?))
    }

    // ── Owner DID device lifecycle ────────────────────────────────────────────

    /// Build the JSON body for `POST /api/v2/did-devices/operations/add/confirm`.
    ///
    /// Extracted so the body shape can be unit-tested without a live daemon.
    /// Must contain ONLY the target `device_id` — never `current_device_id`,
    /// `token`, or any approver identity, since the server derives the
    /// approving device from the authenticated JWT.
    pub(crate) fn confirm_did_device_body(device_id: &str) -> Value {
        json!({ "device_id": device_id })
    }

    /// Extract the `devices` array from a `GET /api/v2/did-devices` response,
    /// tolerating both the legacy envelope and the v2 direct shape.
    pub(crate) fn did_devices_from_response(value: Value) -> Result<Vec<Value>> {
        let data = extract_data(value);
        data.get("devices")
            .and_then(Value::as_array)
            .cloned()
            .ok_or_else(|| anyhow!("device list response missing 'devices' array; got: {data}"))
    }

    /// GET /api/v2/did-devices — list devices registered to the owner's DID.
    pub fn list_did_devices(&self, token: &str) -> Result<Vec<Value>> {
        let value = self.authed_get("/api/v2/did-devices", token)?;
        Self::did_devices_from_response(value)
    }

    /// POST /api/v2/did-devices/operations/add/confirm — approve a pending
    /// device addition. The approving device is derived server-side from the
    /// authenticated JWT; the body carries only the target `device_id`.
    pub fn confirm_did_device(&self, token: &str, device_id: &str) -> Result<Value> {
        Ok(extract_data(self.authed_post(
            "/api/v2/did-devices/operations/add/confirm",
            token,
            &Self::confirm_did_device_body(device_id),
        )?))
    }

    /// POST `{base_url}{route}` with an arbitrary JSON body — used for the L402
    /// cross-node probe, where `from`'s daemon receives the call and its
    /// L402HttpClient proxies it to `to`.
    ///
    /// Returns `(http_status_code, response_body_as_value)`.
    /// Unlike `post_json`, a 402 response is NOT treated as an error — it is
    /// returned to the caller as a soft outcome so the probe can report it.
    /// Deliberately does NOT go through [`Self::authed_post`]'s refresh path:
    /// a 401 here is itself part of the status code this probe reports, and
    /// silently retrying would hide that from the L402 route's caller.
    pub fn post_raw_with_status(
        &self,
        token: &str,
        route: &str,
        body: &Value,
    ) -> Result<(u16, Value)> {
        let url = format!("{}{}", self.base_url, route);
        let request = self
            .agent
            .post(&url)
            .set("Content-Type", "application/json")
            .set("Authorization", &format!("Bearer {token}"));
        match request.send_json(body.clone()) {
            Ok(r) => {
                let status = r.status();
                let value: Value = r
                    .into_json()
                    .unwrap_or_else(|_| json!({}));
                Ok((status, value))
            }
            Err(ureq::Error::Status(code, r)) => {
                // 402 is a soft outcome for the L402 probe; return it rather
                // than bailing so callers can inspect the WWW-Authenticate
                // challenge or print a structured hint.
                let body_str = r.into_string().unwrap_or_default();
                let value: Value = serde_json::from_str(&body_str)
                    .unwrap_or_else(|_| json!({ "raw": body_str }));
                Ok((code, value))
            }
            Err(e) => anyhow::bail!("POST {url} transport error: {e}"),
        }
    }

    // ── Authed request plumbing + 401 refresh (spec D5) ───────────────────────

    /// The freshest known token: the caller-supplied one, unless this client
    /// already recovered a newer one via a prior refresh this process.
    fn effective_token(&self, caller_token: &str) -> String {
        self.current_token
            .lock()
            .unwrap()
            .clone()
            .unwrap_or_else(|| caller_token.to_string())
    }

    fn raw_get(&self, method: &'static str, path: &str, token: &str) -> Result<Value, ProbeError> {
        let response = self
            .agent
            .get(&format!("{}{}", self.base_url, path))
            .set("Authorization", &format!("Bearer {token}"))
            .call();
        Self::classify_response(method, path, response)
    }

    fn raw_post(
        &self,
        method: &'static str,
        path: &str,
        token: &str,
        body: &Value,
    ) -> Result<Value, ProbeError> {
        let response = self
            .agent
            .post(&format!("{}{}", self.base_url, path))
            .set("Content-Type", "application/json")
            .set("Authorization", &format!("Bearer {token}"))
            .send_json(body.clone());
        Self::classify_response(method, path, response)
    }

    fn classify_response(
        method: &'static str,
        path: &str,
        response: std::result::Result<ureq::Response, ureq::Error>,
    ) -> Result<Value, ProbeError> {
        match response {
            Ok(r) => r.into_json::<Value>().map_err(|e| ProbeError::Protocol {
                method,
                path: path.to_string(),
                detail: e.to_string(),
            }),
            Err(ureq::Error::Status(401, r)) => Err(ProbeError::Unauthorized {
                method,
                path: path.to_string(),
                detail: r.into_string().unwrap_or_default(),
            }),
            Err(ureq::Error::Status(status, r)) => Err(ProbeError::Http {
                method,
                path: path.to_string(),
                status,
                body: r.into_string().unwrap_or_default(),
            }),
            Err(ureq::Error::Transport(t)) => Err(ProbeError::Unreachable {
                method,
                path: path.to_string(),
                detail: t.to_string(),
            }),
        }
    }

    /// Run an authed GET, transparently recovering a single 401 — see the
    /// module docs.
    fn authed_get(&self, path: &str, token: &str) -> Result<Value> {
        self.with_refresh(token, |t| self.raw_get("GET", path, t))
    }

    /// Run an authed POST, transparently recovering a single 401 — see the
    /// module docs.
    fn authed_post(&self, path: &str, token: &str, body: &Value) -> Result<Value> {
        self.with_refresh(token, |t| self.raw_post("POST", path, t, body))
    }

    /// Shared retry-once-after-refresh logic for [`Self::authed_get`] /
    /// [`Self::authed_post`]. `attempt` must be safe to call twice (it must
    /// not consume anything by value) since it is invoked once with the
    /// current token and, only on a 401, once more with a refreshed one.
    fn with_refresh(
        &self,
        token: &str,
        attempt: impl Fn(&str) -> Result<Value, ProbeError>,
    ) -> Result<Value> {
        let effective = self.effective_token(token);
        match attempt(&effective) {
            Ok(v) => Ok(v),
            Err(ProbeError::Unauthorized { method, path, detail }) => {
                match self.refresh_and_persist() {
                    Ok(new_token) => match attempt(&new_token) {
                        Ok(v) => Ok(v),
                        Err(ProbeError::Unauthorized { method, path, detail: second_detail }) => {
                            Err(anyhow::Error::new(ProbeError::Unauthorized {
                                method,
                                path,
                                detail: format!(
                                    "still unauthorized after refreshing the access token — the \
                                     refresh token itself may be invalid/expired/revoked. \
                                     Response: {second_detail}"
                                ),
                            }))
                        }
                        Err(other) => Err(anyhow::Error::new(other)),
                    },
                    Err(refresh_err) => Err(anyhow::Error::new(ProbeError::Unauthorized {
                        method,
                        path,
                        detail: format!(
                            "token rejected (401: {detail}), and refreshing it failed: {refresh_err:#}"
                        ),
                    })),
                }
            }
            Err(other) => Err(anyhow::Error::new(other)),
        }
    }

    /// Reload this client's bound session, exchange its `refresh_token` at
    /// `POST /api/auth/refresh`, persist the new token pair via
    /// `AgentSession::save`, cache the new access token for subsequent
    /// calls on this client, and return it.
    ///
    /// Errors when this client has no bound session ([`Self::new`] rather
    /// than [`Self::with_session`]) — such a client has nowhere to read a
    /// `refresh_token` from or to write the result back to.
    fn refresh_and_persist(&self) -> Result<String> {
        let session_path = self.session_path.as_ref().ok_or_else(|| {
            anyhow!(
                "received 401 and this client has no bound session file to refresh from \
                 (constructed via AgentHttpClient::new, not with_session)"
            )
        })?;
        let mut session =
            crate::commands::dev::agent::session::AgentSession::load_from_path(session_path)
                .with_context(|| format!("load session for refresh at {}", session_path.display()))?;
        let refreshed = match self.refresh(&session.refresh_token) {
            Ok(refreshed) => refreshed,
            // The minimal core (#2939) serves no refresh route. A dev
            // instance's JWT secret is in its own daemon.env, so the harness
            // re-issues the same owner token with a fresh expiry instead —
            // the session a browser pairing would otherwise get only by
            // re-onboarding a node that already has an owner.
            Err(server_refresh) => AuthSession {
                token: reissue_from_instance_secret(session_path, &session.token).with_context(|| {
                    format!(
                        "POST /api/auth/refresh failed ({server_refresh:#}), and re-issuing the \
                         owner token from the instance's JWT_SECRET failed too"
                    )
                })?,
                refresh_token: session.refresh_token.clone(),
            },
        };
        session.token = refreshed.token.clone();
        session.refresh_token = refreshed.refresh_token.clone();
        session.last_login_at = chrono::Utc::now();
        let dev_dir = session_path.parent().ok_or_else(|| {
            anyhow!("session path {} has no parent directory", session_path.display())
        })?;
        session
            .save(dev_dir)
            .with_context(|| format!("persist refreshed session to {}", dev_dir.display()))?;
        *self.current_token.lock().unwrap() = Some(refreshed.token.clone());
        Ok(refreshed.token)
    }
}

/// The owner token `expired` re-issued with a fresh `iat`/`exp`, signed with
/// the dev instance's own `JWT_SECRET`.
///
/// The secret is read from the instance's `daemon.env` — `<instance
/// root>/daemon.env`, the parent of the session file's `dev-apps/` dir — and,
/// like `node-server`'s own env loader, the FIRST line naming it wins. The
/// old token's signature is verified with that secret before anything is
/// reused (only its expiry is ignored), so this can only extend a token this
/// instance issued, never mint a new identity.
///
/// Dev instances only: nothing outside a harness/dev cache dir has a
/// `daemon.env` beside its session file.
pub(crate) fn reissue_from_instance_secret(
    session_path: &std::path::Path,
    expired: &str,
) -> Result<String> {
    let env_file = session_path
        .parent()
        .and_then(|dev_dir| dev_dir.parent())
        .map(|root| root.join("daemon.env"))
        .ok_or_else(|| anyhow!("session path {} has no instance root", session_path.display()))?;
    let contents = std::fs::read_to_string(&env_file)
        .with_context(|| format!("read {}", env_file.display()))?;
    let secret = first_env_value(&contents, "JWT_SECRET")
        .ok_or_else(|| anyhow!("{} sets no JWT_SECRET", env_file.display()))?;
    reissue_token(expired, &secret, chrono::Utc::now().timestamp())
}

/// How long a re-issued owner token lives — the minimal core's own access
/// token expiry (`system/server/src/seams.rs`, `jwt_token_expiry: 900`).
const REISSUED_TOKEN_TTL_SECS: i64 = 900;

fn reissue_token(expired: &str, secret: &str, now: i64) -> Result<String> {
    use jsonwebtoken::{decode, encode, Algorithm, DecodingKey, EncodingKey, Header, Validation};
    let mut validation = Validation::new(Algorithm::HS256);
    validation.validate_exp = false;
    validation.required_spec_claims.clear();
    let mut claims = decode::<serde_json::Map<String, Value>>(
        expired,
        &DecodingKey::from_secret(secret.as_bytes()),
        &validation,
    )
    .context("the cached owner token does not verify under this instance's JWT_SECRET")?
    .claims;
    claims.insert("iat".into(), json!(now));
    claims.insert("exp".into(), json!(now + REISSUED_TOKEN_TTL_SECS));
    encode(
        &Header::new(Algorithm::HS256),
        &claims,
        &EncodingKey::from_secret(secret.as_bytes()),
    )
    .context("sign the re-issued owner token")
}

/// `KEY=value` as `node-server`'s env loader reads it: first occurrence wins,
/// an `export ` prefix and one layer of matching quotes are stripped.
fn first_env_value(contents: &str, wanted: &str) -> Option<String> {
    contents.lines().find_map(|line| {
        let line = line.trim();
        if line.starts_with('#') {
            return None;
        }
        let (key, value) = line.split_once('=')?;
        let key = key.trim();
        let key = key.strip_prefix("export ").unwrap_or(key);
        if key != wanted {
            return None;
        }
        let value = value.trim();
        let value = value
            .strip_prefix('"')
            .and_then(|v| v.strip_suffix('"'))
            .or_else(|| value.strip_prefix('\'').and_then(|v| v.strip_suffix('\'')))
            .unwrap_or(value);
        Some(value.to_string()).filter(|v| !v.is_empty())
    })
}

fn post_json(
    agent: &ureq::Agent,
    url: &str,
    bearer: Option<&str>,
    body: &Value,
) -> Result<Value> {
    let mut request = agent.post(url).set("Content-Type", "application/json");
    if let Some(token) = bearer {
        request = request.set("Authorization", &format!("Bearer {token}"));
    }
    let response = match request.send_json(body.clone()) {
        Ok(r) => r,
        Err(ureq::Error::Status(code, r)) => {
            let body = r.into_string().unwrap_or_default();
            bail!("POST {url} returned HTTP {code}: {body}");
        }
        Err(e) => bail!("POST {url} transport error: {e}"),
    };
    response
        .into_json::<Value>()
        .with_context(|| format!("parse JSON response from POST {url}"))
}

/// Accept both shapes the daemon returns:
///   { "success": true, "data": {...} }       (legacy v1 envelope)
///   {...}                                    (v2 hexagonal direct return)
fn extract_data(value: Value) -> Value {
    match value {
        Value::Object(ref obj) if obj.contains_key("success") && obj.contains_key("data") => {
            obj.get("data").cloned().unwrap_or(Value::Null)
        }
        other => other,
    }
}

fn string_field(value: &Value, key: &str) -> Result<String> {
    value
        .get(key)
        .and_then(Value::as_str)
        .map(str::to_owned)
        .ok_or_else(|| anyhow!("response missing '{key}' string field; got: {value}"))
}

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

    #[test]
    fn open_channel_body_has_expected_fields() {
        let body = AgentHttpClient::open_channel_body("03aa@127.0.0.1:9536", 100_000, 10_000_000);
        assert_eq!(body["peer_pubkey_and_address"], "03aa@127.0.0.1:9536");
        assert_eq!(body["channel_amount_sats"], 100_000u64);
        assert_eq!(body["push_to_counterparty_msat"], 10_000_000u64);
    }

    #[test]
    fn confirm_did_device_body_contains_only_target_device() {
        let body = AgentHttpClient::confirm_did_device_body("browser-device");
        assert_eq!(body, json!({ "device_id": "browser-device" }));
        assert!(body.get("current_device_id").is_none());
        assert!(body.get("token").is_none());
    }

    #[test]
    fn did_device_list_accepts_enveloped_shape() {
        let devices = AgentHttpClient::did_devices_from_response(json!({
            "success": true,
            "data": {
                "devices": [
                    {
                        "device_id": "browser-device",
                        "status": "pending",
                        "created_at": "2026-07-25T00:00:01Z"
                    }
                ]
            }
        }))
        .unwrap();
        assert_eq!(devices[0]["device_id"], "browser-device");
    }

    #[test]
    fn did_device_list_rejects_missing_devices_array() {
        let error = AgentHttpClient::did_devices_from_response(json!({
            "success": true,
            "data": {}
        }))
        .unwrap_err();
        assert!(error.to_string().contains("devices"));
    }
}

#[cfg(test)]
mod reissue_tests {
    use super::*;
    use jsonwebtoken::{decode, encode, Algorithm, DecodingKey, EncodingKey, Header, Validation};

    fn owner_token(secret: &str, iat: i64, exp: i64) -> String {
        encode(
            &Header::new(Algorithm::HS256),
            &json!({
                "sub": "user-1", "node_id": "02ab", "public_key_hash": "h",
                "public_key": "04cd", "device_id": "d-1", "iat": iat, "exp": exp,
                "is_primary_owner": true, "token_type": "access",
            }),
            &EncodingKey::from_secret(secret.as_bytes()),
        )
        .unwrap()
    }

    #[test]
    fn a_reissued_token_keeps_every_claim_and_gets_a_fresh_expiry() {
        // The minimal core has no /api/auth/refresh, so an owner token older
        // than 15 minutes could only be replaced by re-onboarding a node that
        // already has an owner — which it refuses with 409.
        let expired = owner_token("s3cret", 1_000, 1_900);
        let fresh = reissue_token(&expired, "s3cret", 50_000).unwrap();
        // `now` is pinned in the past, so check the expiry arithmetic
        // explicitly rather than against the real clock.
        let mut validation = Validation::new(Algorithm::HS256);
        validation.validate_exp = false;
        let claims = decode::<Value>(&fresh, &DecodingKey::from_secret(b"s3cret"), &validation)
            .unwrap()
            .claims;
        assert_eq!(claims["iat"], 50_000);
        assert_eq!(claims["exp"], 50_000 + REISSUED_TOKEN_TTL_SECS);
        for (key, value) in [("sub", "user-1"), ("device_id", "d-1"), ("token_type", "access")] {
            assert_eq!(claims[key], value, "{key}");
        }
        assert_eq!(claims["is_primary_owner"], true);
    }

    #[test]
    fn a_token_signed_with_another_secret_is_never_reissued() {
        let foreign = owner_token("someone-else", 1_000, 1_900);
        assert!(reissue_token(&foreign, "s3cret", 50_000).is_err());
    }

    #[test]
    fn the_secret_is_read_the_way_node_server_reads_its_env_file() {
        let contents = "# JWT_SECRET=commented\nexport JWT_SECRET=\"first\"\nJWT_SECRET=second\n";
        assert_eq!(first_env_value(contents, "JWT_SECRET").as_deref(), Some("first"));
        assert_eq!(first_env_value("JWT_SECRET=\n", "JWT_SECRET"), None);
        assert_eq!(first_env_value("OTHER=1\n", "JWT_SECRET"), None);
    }

    #[test]
    fn reissue_reads_the_secret_from_the_instance_daemon_env() {
        let root = tempfile::tempdir().unwrap();
        std::fs::create_dir_all(root.path().join("dev-apps")).unwrap();
        std::fs::write(root.path().join("daemon.env"), "JWT_SECRET=abc\nJWT_SECRET=ignored\n")
            .unwrap();
        let session = root.path().join("dev-apps/alice-agent-session.json");
        let expired = owner_token("abc", 1_000, 1_900);
        assert!(reissue_from_instance_secret(&session, &expired).is_ok());
        let wrong = owner_token("ignored", 1_000, 1_900);
        assert!(reissue_from_instance_secret(&session, &wrong).is_err());
    }
}

#[cfg(test)]
mod refresh_tests {
    //! Exercises the real `ureq` client against a hand-rolled HTTP/1.1 stub
    //! (no live daemon, no extra test-only HTTP dependency) to prove the
    //! 401 → refresh → retry-once behavior end to end, including that the
    //! new token pair actually lands on disk via `AgentSession::save`.

    use super::*;
    use crate::commands::dev::agent::session::AgentSession;
    use std::io::{Read, Write};
    use std::net::TcpListener;
    use std::sync::mpsc;
    use std::time::Duration;

    /// One scripted HTTP/1.1 response: fixed status + JSON body.
    struct Canned {
        status: u16,
        body: String,
    }

    fn canned(status: u16, body: Value) -> Canned {
        Canned { status, body: body.to_string() }
    }

    fn status_reason(status: u16) -> &'static str {
        match status {
            200 => "OK",
            401 => "Unauthorized",
            _ => "Status",
        }
    }

    /// How [`spawn_stub`]'s background thread finished: either it served
    /// every scripted response, or its own deadline elapsed first — the
    /// latter reported with how many it got through, never left silent.
    enum StubOutcome {
        ServedAll,
        GaveUpAfter { served: usize, expected: usize },
    }

    /// Bound on the stub's own accept-loop, independent of anything the
    /// client does. Exists so a client that — for whatever reason, under
    /// whatever load — makes fewer connections than scripted (e.g. an
    /// `AgentHttpClient` returning after 2 of an expected 3 requests)
    /// produces a REPORTED, bounded outcome instead of leaving the stub
    /// thread blocked in `accept()` forever, which previously hung this
    /// exact test for minutes under real box contention (reproduced live,
    /// not hypothetical) with nothing to distinguish it from a true deadlock.
    const STUB_DEADLINE: Duration = Duration::from_secs(10);

    /// Minimal single-purpose HTTP/1.1 stub: accepts up to `responses.len()`
    /// connections, in order, and writes back the corresponding canned
    /// response to each — but never blocks past [`STUB_DEADLINE`] waiting
    /// for a connection that doesn't arrive. Reports its outcome on `done`
    /// rather than via a plain `JoinHandle`, specifically so callers can
    /// bound their own wait with `recv_timeout` instead of an unboundable
    /// `JoinHandle::join()`.
    ///
    /// Good enough for these tests' small single-shot JSON bodies — no
    /// chunked encoding, no keep-alive (`Connection: close` forces the
    /// client to open a fresh connection per request, which is what makes
    /// "in order" a valid assumption when it does complete normally).
    fn spawn_stub(responses: Vec<Canned>) -> (String, mpsc::Receiver<StubOutcome>) {
        let listener = TcpListener::bind("127.0.0.1:0").expect("bind stub listener");
        listener
            .set_nonblocking(true)
            .expect("set stub listener non-blocking (needed to bound accept())");
        let addr = listener.local_addr().expect("stub listener addr");
        let (tx, rx) = mpsc::channel();
        std::thread::spawn(move || {
            let expected = responses.len();
            let deadline = std::time::Instant::now() + STUB_DEADLINE;
            for (served, canned) in responses.into_iter().enumerate() {
                let mut stream = loop {
                    match listener.accept() {
                        Ok((stream, _)) => break stream,
                        Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                            if std::time::Instant::now() >= deadline {
                                let _ = tx.send(StubOutcome::GaveUpAfter { served, expected });
                                return;
                            }
                            std::thread::sleep(Duration::from_millis(20));
                        }
                        Err(_) => {
                            let _ = tx.send(StubOutcome::GaveUpAfter { served, expected });
                            return;
                        }
                    }
                };
                // Blocking mode is simpler for the tiny synchronous
                // read/write below; only `accept()` needed to be pollable.
                let _ = stream.set_nonblocking(false);
                let _ = stream.set_read_timeout(Some(Duration::from_secs(5)));
                let mut buf = [0u8; 8192];
                let _ = stream.read(&mut buf); // drain the request; bodies here are tiny.
                let response = format!(
                    "HTTP/1.1 {} {}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                    canned.status,
                    status_reason(canned.status),
                    canned.body.len(),
                    canned.body,
                );
                let _ = stream.write_all(response.as_bytes());
                let _ = stream.flush();
            }
            let _ = tx.send(StubOutcome::ServedAll);
        });
        (format!("http://{addr}"), rx)
    }

    /// Wait for the stub's outcome with an EXTRA bound on top of its own
    /// [`STUB_DEADLINE`] (covers the thread-scheduling delay between the
    /// stub finishing and this end actually observing it) — this call can
    /// never hang the test, unlike a bare `JoinHandle::join()`. Returns
    /// `Err` (a transient-flake-shaped message, not a panic) whenever the
    /// stub did not serve every scripted response, so callers can route it
    /// through [`retry_transient_env_flake`] uniformly with a client-side
    /// failure instead of needing a separate code path.
    fn await_stub(rx: &mpsc::Receiver<StubOutcome>) -> Result<(), String> {
        match rx.recv_timeout(STUB_DEADLINE + Duration::from_secs(2)) {
            Ok(StubOutcome::ServedAll) => Ok(()),
            Ok(StubOutcome::GaveUpAfter { served, expected }) => Err(format!(
                "stub server only saw {served}/{expected} scripted connections before its own \
                 {STUB_DEADLINE:?} deadline — the client made fewer requests than the scenario \
                 scripted (transient loopback flake or the client returned early)"
            )),
            Err(_) => Err(format!(
                "stub server thread never reported an outcome within {:?}",
                STUB_DEADLINE + Duration::from_secs(2)
            )),
        }
    }

    fn write_test_session(dir: &std::path::Path, token: &str, refresh_token: &str, base_url: &str) -> AgentSession {
        let session = AgentSession {
            instance: "alice".into(),
            base_url: base_url.into(),
            node_id: "03aa".into(),
            public_key: "pub".into(),
            secret_key_hex: "sec".into(),
            mnemonic: "test mnemonic".into(),
            token: token.into(),
            refresh_token: refresh_token.into(),
            onboarded_at: chrono::Utc::now(),
            last_login_at: chrono::Utc::now(),
        };
        session.save(dir).expect("write test session");
        session
    }

    /// This box runs several concurrent sessions (worktrees, other agents'
    /// test suites), and these two tests each open TWO real loopback TCP
    /// connections back to back (the initial 401 GET, then the refresh
    /// POST). Under heavy parallel load — reliably reproduced with `cargo
    /// test`'s default thread-per-core scheduling, never with
    /// `--test-threads=1` — that second connection occasionally trips a
    /// transient OS-level error unrelated to the refresh logic under test
    /// (surfaces as a `ProbeError::Unauthorized` whose detail names a raw
    /// transport error rather than "still unauthorized after refreshing").
    /// `scenario` performs its own setup AND its own assertions and should
    /// return `Err` ONLY to report that specific environmental shape —
    /// every other failure (a wrong value, a wrong error shape) is a real
    /// bug and panics immediately, unswallowed by this retry.
    fn retry_transient_env_flake(mut scenario: impl FnMut() -> Result<(), String>) {
        for attempt in 1..=3 {
            match scenario() {
                Ok(()) => return,
                Err(msg) if attempt < 3 => {
                    eprintln!(
                        "refresh_tests: retrying after a transient loopback-socket flake \
                         (attempt {attempt}/3; shared box under load, not a logic bug): {msg}"
                    );
                }
                Err(msg) => panic!("giving up after {attempt} attempts: {msg}"),
            }
        }
    }

    /// True only for the specific "the refresh round trip itself hit a raw
    /// transport error" shape — never for "still unauthorized after
    /// refreshing", which is the real behavior under test in
    /// `a_second_401_after_refresh_is_reported_as_unauthorized_not_generic`
    /// and must keep failing the test immediately if it regresses.
    fn is_transient_env_flake(err: &anyhow::Error) -> bool {
        let msg = format!("{err:#}");
        (msg.contains("os error") || msg.contains("Network Error") || msg.contains("transport error"))
            && !msg.contains("still unauthorized after refreshing")
    }

    #[test]
    fn a_401_triggers_refresh_and_retries_once() {
        retry_transient_env_flake(|| {
            let dir = tempfile::tempdir().expect("tempdir");
            // The 401's response body deliberately mirrors the real daemon's
            // (middleware.rs:152), so this test also documents what a caller
            // actually sees.
            let (base_url, done_rx) = spawn_stub(vec![
                canned(401, json!({ "error": "Invalid or expired token" })),
                canned(200, json!({ "token": "new-access-token", "refresh_token": "new-refresh-token" })),
                canned(200, json!({ "total_onchain_balance_sats": 42 })),
            ]);

            let session = write_test_session(dir.path(), "stale-token", "still-valid-refresh", &base_url);
            let session_path = AgentSession::file_path(dir.path(), &session.instance);

            let client = AgentHttpClient::with_session(base_url, session_path.clone());
            let result = client.get_balance("stale-token");
            let stub_result = await_stub(&done_rx);

            let balance = match result {
                Ok(v) => v,
                Err(e) if is_transient_env_flake(&e) => return Err(e.to_string()),
                Err(e) => panic!("get_balance should recover via refresh: {e:#}"),
            };
            // The client reported success — the stub must have genuinely
            // served all 3 scripted responses, not given up early.
            stub_result?;
            assert_eq!(balance["total_onchain_balance_sats"], 42);

            // The refreshed token pair must actually be on disk — this is the
            // second defect in spec D5 (refresh_token was written and never
            // read), so a test that only checks the in-memory return value
            // would miss it entirely.
            let persisted = AgentSession::load_from_path(&session_path).expect("reload session");
            assert_eq!(persisted.token, "new-access-token");
            assert_eq!(persisted.refresh_token, "new-refresh-token");
            Ok(())
        });
    }

    #[test]
    fn a_second_401_after_refresh_is_reported_as_unauthorized_not_generic() {
        retry_transient_env_flake(|| {
            let dir = tempfile::tempdir().expect("tempdir");
            let (base_url, done_rx) = spawn_stub(vec![
                canned(401, json!({ "error": "Invalid or expired token" })),
                canned(200, json!({ "token": "new-access-token", "refresh_token": "new-refresh-token" })),
                canned(401, json!({ "error": "Invalid or expired token" })),
            ]);

            let session = write_test_session(dir.path(), "stale-token", "dead-refresh-token", &base_url);
            let session_path = AgentSession::file_path(dir.path(), &session.instance);
            let client = AgentHttpClient::with_session(base_url, session_path);

            let result = client.get_balance("stale-token");
            let stub_result = await_stub(&done_rx);

            let error = match result {
                Err(e) => e,
                Ok(v) => panic!("expected the second 401 to surface as an error, got: {v}"),
            };
            if is_transient_env_flake(&error) {
                return Err(error.to_string());
            }
            // The client reported the expected shape of error — confirm the
            // stub actually ran the full scripted exchange rather than
            // giving up early (which could otherwise coincidentally produce
            // an error that LOOKS like the right shape for the wrong reason).
            stub_result?;
            let probe_error = error
                .downcast_ref::<ProbeError>()
                .unwrap_or_else(|| panic!("error must be a classified ProbeError, not an opaque string: {error:#}"));
            assert!(
                matches!(probe_error, ProbeError::Unauthorized { .. }),
                "still unauthorized after a successful refresh must stay classified as Unauthorized: {probe_error:?}"
            );
            assert!(
                error.to_string().contains("still unauthorized after refreshing"),
                "message must say the refresh itself didn't help: {error}"
            );
            Ok(())
        });
    }

    #[test]
    fn a_401_with_no_bound_session_cannot_refresh_and_says_so() {
        let (base_url, done_rx) = spawn_stub(vec![canned(401, json!({ "error": "Invalid or expired token" }))]);

        // Plain `new`, not `with_session` — no session file to refresh from.
        let client = AgentHttpClient::new(base_url);
        let error = client.get_balance("stale-token").unwrap_err();
        assert!(
            error.downcast_ref::<ProbeError>().is_some(),
            "even the unrecoverable case must stay a classified ProbeError"
        );
        assert!(
            error.to_string().contains("no bound session"),
            "message must explain why refresh could not be attempted: {error}"
        );

        await_stub(&done_rx).expect("stub should have served its single scripted response");
    }

    #[test]
    fn an_unreachable_daemon_is_classified_distinctly_from_unauthorized() {
        // Nothing is listening on this port — a real "unreachable" case,
        // not a stub. Using a stub for a 401 above and a real closed port
        // here is what actually proves the two are told apart: they go
        // through genuinely different ureq::Error variants.
        let listener = TcpListener::bind("127.0.0.1:0").expect("bind to find a free port");
        let addr = listener.local_addr().expect("addr");
        drop(listener); // free the port; nothing will accept on it now.

        let client = AgentHttpClient::new(format!("http://{addr}"));
        let error = client.get_balance("token").unwrap_err();
        let probe_error = error
            .downcast_ref::<ProbeError>()
            .expect("must be a classified ProbeError");
        assert!(
            matches!(probe_error, ProbeError::Unreachable { .. }),
            "a connection failure must not be misreported as an auth failure: {probe_error:?}"
        );
    }
}