zc2 0.0.29

//! Broker-to-broker P2P communication for credit operations.
//!
//! Each user is assigned to exactly one broker as the "authoritative credit owner"
//! (determined by `hash(user_id) % num_brokers`). The authoritative broker manages
//! that user's balance in-memory. Non-authoritative brokers forward credit operations
//! to the authoritative peer via lightweight HTTP calls over WireGuard.

use dashmap::DashMap;
use serde::{Deserialize, Serialize};
use std::collections::BTreeSet;
use std::net::{SocketAddr, ToSocketAddrs};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;

use super::node_identity::NodeKey;

// --- Request / Response types ---

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PeerReserveRequest {
    pub user_id: String,
    pub amount: f64,
    pub request_id: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PeerReserveResponse {
    pub reservation_id: String,
    pub balance_before: f64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PeerCommitRequest {
    pub reservation_id: String,
    pub actual_cost: f64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PeerCommitResponse {
    pub balance_after: f64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PeerCancelRequest {
    pub reservation_id: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PeerEarnRequest {
    /// Duration-based credits earned (without min_charge floor)
    pub amount: f64,
    /// Actual wall-clock duration of the job in milliseconds
    pub duration_ms: f64,
    /// Worker ID that performed the job
    pub worker_id: String,
    /// User who paid for the compute (for audit trail)
    pub requesting_user: String,
    /// Original request ID (for idempotency / dashboard dedup)
    pub request_id: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PeerBalanceResponse {
    pub user_id: String,
    pub balance: f64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PeerHealthResponse {
    pub status: String,
    pub broker_id: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PeerErrorResponse {
    pub error: String,
    pub code: String,
}

// --- Authority enum ---

/// Determines which broker is authoritative for a given user.
#[derive(Debug, Clone)]
pub enum Authority {
    /// This broker is authoritative — use local in-memory operations.
    Local,
    /// A peer broker is authoritative — forward credit ops via HTTP.
    Peer(String), // peer base URL, e.g. "http://100.64.0.5:9000"
    /// P2P disabled or peer unreachable — use local in-memory operations.
    Standalone,
}

/// A cached peer entry persisted to `<dir>/peers.json` so the peer set can
/// be restored across broker restarts.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PersistedPeer {
    pub fp: String,
    pub url: String,
    pub last_seen: u64,
    /// Whether this peer was learned from the dashboard roster (evictable)
    /// rather than configured via `ZAKURO_PEERS` (permanent).
    ///
    /// `#[serde(default)]` is load-bearing twice over: a `peers.json` written
    /// by an older zc has no such key, and defaulting to `false` means those
    /// entries reload as CONFIGURED — i.e. never evicted, exactly today's
    /// behaviour — so an upgrade cannot silently start dropping a peer the
    /// fleet depends on. Without persisting this flag, a roster-derived peer
    /// would come back after a restart as an un-evictable configured one and
    /// re-open the leak this field exists to close.
    #[serde(default)]
    pub roster_derived: bool,
}

// --- PeerClient ---

/// The query-less request path a signature covers, extracted from a full URL.
/// `http://10.13.13.5:9000/peer/balance?user_id=1` → `/peer/balance`.
fn url_sign_path(url: &str) -> &str {
    let after_scheme = url.split("://").nth(1).unwrap_or(url);
    let path = after_scheme
        .find('/')
        .map(|i| &after_scheme[i..])
        .unwrap_or("/");
    path.split('?').next().unwrap_or(path)
}

/// HTTP client for calling peer broker endpoints. Uses ureq::Agent for
/// keep-alive connection reuse.
pub struct PeerClient {
    agent: ureq::Agent,
    base_url: String,
    peer_key: String,
    reachable: AtomicBool,
    /// This node's signing identity — attaches X-Node-* to every peer request
    /// (zero-trust). `None` only in tests / keyless construction.
    node_key: Option<Arc<NodeKey>>,
    /// Peer's key-derived identity (`zc://node-<fp>`), learned from the
    /// `broker_id` field of a successful `/peer/health` probe and cached
    /// here — IP-free, so `/brokers` can read it back without ever touching
    /// `base_url` (which carries the peer's IP).
    fingerprint: std::sync::Mutex<Option<String>>,
    /// This peer's most recently fetched-and-verified advertisement (price +
    /// aggregated resources), cached via `PeerManager::set_peer_advert`.
    /// `None` until the first successful `discovery::fetch_advert`.
    advert: std::sync::Mutex<Option<super::discovery::Advert>>,
    /// Local wall-clock seconds when `advert` was last set (freshness stamp
    /// for the cache, independent of the advert's own `epoch`).
    advert_seen_at: std::sync::atomic::AtomicU64,
    /// Round-trip time (ms) of the most recent `/peer/health` probe
    /// (`check_health`), stored as bits of an f64 for lock-free access.
    /// `u64::MAX` sentinel = "never successfully probed" (see
    /// `last_rtt_ms()` — never surfaced as a real latency).
    last_rtt_ms_bits: AtomicU64,
}

impl PeerClient {
    /// Signed `X-Node-*` headers for one outbound request (empty when no key).
    /// `path` MUST be the query-less path the server matches on (e.g. `/peer/balance`).
    fn node_headers(&self, method: &str, path: &str, body: &[u8]) -> Vec<(String, String)> {
        match &self.node_key {
            Some(k) => k.sign_headers(method, path, body),
            None => Vec::new(),
        }
    }

    /// Signed GET to a peer endpoint (sign path derived from the URL).
    fn signed_get(&self, url: &str) -> Result<ureq::http::Response<ureq::Body>, ureq::Error> {
        let mut rb = self.agent.get(url).header("X-Peer-Key", &self.peer_key);
        for (k, v) in self.node_headers("GET", url_sign_path(url), b"") {
            rb = rb.header(k.as_str(), v.as_str());
        }
        rb.call()
    }

    /// Signed JSON POST to a peer endpoint (sign path derived from the URL).
    fn signed_post(
        &self,
        url: &str,
        body: &[u8],
    ) -> Result<ureq::http::Response<ureq::Body>, ureq::Error> {
        let mut rb = self
            .agent
            .post(url)
            .header("X-Peer-Key", &self.peer_key)
            .header("Content-Type", "application/json");
        for (k, v) in self.node_headers("POST", url_sign_path(url), body) {
            rb = rb.header(k.as_str(), v.as_str());
        }
        rb.send(body)
    }

    /// Signed async (reqwest) JSON POST — returns a builder ready to `.send()`.
    fn signed_reqwest_post(
        &self,
        http: &reqwest::Client,
        url: &str,
        body: Vec<u8>,
    ) -> reqwest::RequestBuilder {
        let mut rb = http
            .post(url)
            .header("X-Peer-Key", &self.peer_key)
            .header("Content-Type", "application/json");
        for (k, v) in self.node_headers("POST", url_sign_path(url), &body) {
            rb = rb.header(k.as_str(), v.as_str());
        }
        rb.body(body)
    }

    /// Signed async (reqwest) GET — returns a builder ready to `.send()`.
    fn signed_reqwest_get(&self, http: &reqwest::Client, url: &str) -> reqwest::RequestBuilder {
        let mut rb = http.get(url).header("X-Peer-Key", &self.peer_key);
        for (k, v) in self.node_headers("GET", url_sign_path(url), b"") {
            rb = rb.header(k.as_str(), v.as_str());
        }
        rb
    }

    pub fn new(base_url: String, peer_key: String) -> Self {
        Self::new_with_node_key(base_url, peer_key, None)
    }

    pub fn new_with_node_key(
        base_url: String,
        peer_key: String,
        node_key: Option<Arc<NodeKey>>,
    ) -> Self {
        let mut cfg = ureq::Agent::config_builder()
            // 2s connect tolerates a cold mesh handshake (WireGuard hub hairpin);
            // once warm, the pool reuses the connection so this is paid once.
            .timeout_connect(Some(Duration::from_millis(2000)))
            .http_status_as_error(false);
        // Route through the vpn sidecar's CONNECT proxy when this peer lives on
        // the WireGuard mesh (10.13.13.0/24) and a proxy is configured.
        let host = base_url
            .split("://")
            .nth(1)
            .and_then(|r| r.split([':', '/']).next())
            .unwrap_or("");
        if crate::vpn::is_mesh_ip(host) {
            if let Some(addr) = crate::vpn::mesh_proxy_addr() {
                if let Ok(proxy) = ureq::Proxy::new(&format!("http://{}", addr)) {
                    cfg = cfg.proxy(Some(proxy));
                }
            }
        }
        let agent = ureq::Agent::new_with_config(cfg.build());

        Self {
            agent,
            base_url,
            peer_key,
            reachable: AtomicBool::new(false),
            node_key,
            fingerprint: std::sync::Mutex::new(None),
            advert: std::sync::Mutex::new(None),
            advert_seen_at: std::sync::atomic::AtomicU64::new(0),
            last_rtt_ms_bits: AtomicU64::new(u64::MAX),
        }
    }

    pub fn base_url(&self) -> &str {
        &self.base_url
    }

    pub fn is_reachable(&self) -> bool {
        self.reachable.load(Ordering::Relaxed)
    }

    /// The peer's key-derived identity, if learned yet. ALWAYS in the canonical
    /// addressable `zc://node-<fp>` form — see [`set_fingerprint`](Self::set_fingerprint),
    /// the single writer that guarantees it. `None` until the identity is known
    /// (or if the peer predates key-derived `broker_id`).
    pub fn fingerprint(&self) -> Option<String> {
        self.fingerprint.lock().ok().and_then(|g| g.clone())
    }

    /// Cache this peer's key-derived identity. Accepts either the bare 16-hex
    /// fingerprint or the `zc://node-<fp>` form and **canonicalizes to the
    /// latter**, so every reader of [`fingerprint`](Self::fingerprint) sees one
    /// shape.
    ///
    /// Canonicalizing here rather than at each call site is deliberate: the
    /// writers are `check_health` (which reports `zc://node-<fp>`) and the two
    /// eager seeding paths in `PeerManager::load_persisted_from` /
    /// `discovery::admit_verified_peer_with` (which hold a bare fp), and
    /// `fingerprint()` is read verbatim into node-facing payloads — `/peers`
    /// and `/brokers`' `BrokerEntry.id`. A mixed cache would emit two different
    /// id shapes for the same kind of peer depending on whether the ~60s health
    /// tick had run yet, handing clients a non-addressable string. Enforcing the
    /// invariant at the single write seam makes any future third writer correct
    /// by construction instead of relying on it remembering to format.
    ///
    /// An empty/prefix-only argument is ignored — a blank identity must stay
    /// `None` rather than becoming the matches-nothing string `zc://node-`.
    pub fn set_fingerprint(&self, fp: &str) {
        let bare = super::node_identity::strip_node_arg(fp);
        if bare.is_empty() {
            return;
        }
        let canonical = format!("zc://node-{bare}");
        if let Ok(mut g) = self.fingerprint.lock() {
            *g = Some(canonical);
        }
    }

    /// This peer's most recently cached advertisement, if any.
    pub fn advert(&self) -> Option<super::discovery::Advert> {
        self.advert.lock().ok().and_then(|g| g.clone())
    }

    /// Cache a freshly fetched-and-verified advertisement for this peer.
    pub fn set_advert(&self, advert: super::discovery::Advert) {
        if let Ok(mut g) = self.advert.lock() {
            *g = Some(advert);
        }
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);
        self.advert_seen_at.store(now, Ordering::Relaxed);
    }

    /// Probe /peer/health to check if peer is alive. On success, also learns
    /// and caches the peer's key-derived identity (`broker_id` in the
    /// response body, expected to be `zc://node-<fp>`) for later IP-free
    /// lookup via `fingerprint()` — this is how `/brokers` resolves a peer's
    /// id without ever exposing `base_url` (which carries the peer's IP) to
    /// a client.
    pub fn check_health(&self) -> bool {
        let url = format!("{}/peer/health", self.base_url);
        let probe_start = std::time::Instant::now();
        let result = self.signed_get(&url);
        let rtt_ms = probe_start.elapsed().as_secs_f64() * 1000.0;
        let ok = match result {
            Ok(resp) if resp.status().as_u16() == 200 => {
                if let Ok(text) = resp.into_body().read_to_string() {
                    if let Ok(health) = serde_json::from_str::<PeerHealthResponse>(&text) {
                        // Route through `set_fingerprint` so this writer shares
                        // the one canonicalization seam with the eager seeding
                        // paths (finding 3) instead of storing its own shape.
                        if health.broker_id.starts_with("zc://node-") {
                            self.set_fingerprint(&health.broker_id);
                        }
                    }
                }
                true
            }
            _ => false,
        };
        self.reachable.store(ok, Ordering::Relaxed);
        // Only record RTT for a successful probe — a failed/timed-out call's
        // elapsed time is not a meaningful latency measurement.
        if ok {
            self.last_rtt_ms_bits
                .store(rtt_ms.to_bits(), Ordering::Relaxed);
        }
        ok
    }

    /// Round-trip time (ms) of the most recent successful `/peer/health`
    /// probe, or `None` if this peer has never been successfully probed.
    pub fn last_rtt_ms(&self) -> Option<f64> {
        let bits = self.last_rtt_ms_bits.load(Ordering::Relaxed);
        if bits == u64::MAX {
            None
        } else {
            Some(f64::from_bits(bits))
        }
    }

    /// Reserve credits on the authoritative peer.
    pub fn reserve(
        &self,
        user_id: &str,
        amount: f64,
        request_id: &str,
    ) -> Result<PeerReserveResponse, String> {
        let url = format!("{}/peer/reserve", self.base_url);
        let req = PeerReserveRequest {
            user_id: user_id.to_string(),
            amount,
            request_id: request_id.to_string(),
        };
        let body = serde_json::to_vec(&req).map_err(|e| e.to_string())?;

        match self.signed_post(&url, &body) {
            Ok(resp) if resp.status().as_u16() == 200 => {
                self.reachable.store(true, Ordering::Relaxed);
                let text = resp
                    .into_body()
                    .read_to_string()
                    .map_err(|e| e.to_string())?;
                serde_json::from_str(&text).map_err(|e| e.to_string())
            }
            Ok(resp) => {
                self.reachable.store(true, Ordering::Relaxed);
                let code = resp.status().as_u16();
                let text = resp.into_body().read_to_string().unwrap_or_default();
                if let Ok(err) = serde_json::from_str::<PeerErrorResponse>(&text) {
                    Err(format!("[{}] {}", code, err.error))
                } else {
                    Err(format!("Peer returned {}: {}", code, text))
                }
            }
            Err(e) => {
                self.reachable.store(false, Ordering::Relaxed);
                Err(format!("Peer unreachable: {}", e))
            }
        }
    }

    /// Commit a reservation on the authoritative peer.
    pub fn commit(
        &self,
        reservation_id: &str,
        actual_cost: f64,
    ) -> Result<PeerCommitResponse, String> {
        let url = format!("{}/peer/commit", self.base_url);
        let req = PeerCommitRequest {
            reservation_id: reservation_id.to_string(),
            actual_cost,
        };
        let body = serde_json::to_vec(&req).map_err(|e| e.to_string())?;

        match self.signed_post(&url, &body) {
            Ok(resp) if resp.status().as_u16() == 200 => {
                self.reachable.store(true, Ordering::Relaxed);
                let text = resp
                    .into_body()
                    .read_to_string()
                    .map_err(|e| e.to_string())?;
                serde_json::from_str(&text).map_err(|e| e.to_string())
            }
            Ok(resp) => {
                self.reachable.store(true, Ordering::Relaxed);
                let code = resp.status().as_u16();
                let text = resp.into_body().read_to_string().unwrap_or_default();
                Err(format!("Peer commit failed ({}): {}", code, text))
            }
            Err(e) => {
                self.reachable.store(false, Ordering::Relaxed);
                Err(format!("Peer unreachable: {}", e))
            }
        }
    }

    /// Cancel a reservation on the authoritative peer.
    pub fn cancel(&self, reservation_id: &str) -> Result<(), String> {
        let url = format!("{}/peer/cancel", self.base_url);
        let req = PeerCancelRequest {
            reservation_id: reservation_id.to_string(),
        };
        let body = serde_json::to_vec(&req).map_err(|e| e.to_string())?;

        match self.signed_post(&url, &body) {
            Ok(resp) => {
                // The peer answered → reachable, regardless of status. But a
                // non-2xx means the cancel was rejected, not applied (audit M2).
                self.reachable.store(true, Ordering::Relaxed);
                let code = resp.status().as_u16();
                if (200..300).contains(&code) {
                    Ok(())
                } else {
                    Err(format!("Peer cancel failed: HTTP {}", code))
                }
            }
            Err(e) => {
                self.reachable.store(false, Ordering::Relaxed);
                Err(format!("Peer unreachable: {}", e))
            }
        }
    }

    /// Notify the peer that their worker earned credits for a completed job.
    /// `amount` is the duration-based earn (without min_charge floor).
    /// `duration_ms` is the actual wall-clock job time (for compute_hours in dashboard).
    /// Best-effort — callers should log but not fail on error.
    pub fn earn(
        &self,
        amount: f64,
        duration_ms: f64,
        worker_id: &str,
        requesting_user: &str,
        request_id: &str,
    ) -> Result<(), String> {
        let url = format!("{}/peer/earn", self.base_url);
        let req = PeerEarnRequest {
            amount,
            duration_ms,
            worker_id: worker_id.to_string(),
            requesting_user: requesting_user.to_string(),
            request_id: request_id.to_string(),
        };
        let body = serde_json::to_vec(&req).map_err(|e| e.to_string())?;

        match self.signed_post(&url, &body) {
            Ok(resp) => {
                self.reachable.store(true, Ordering::Relaxed);
                let code = resp.status().as_u16();
                if (200..300).contains(&code) {
                    Ok(())
                } else {
                    Err(format!("Peer earn failed: HTTP {}", code))
                }
            }
            Err(e) => {
                self.reachable.store(false, Ordering::Relaxed);
                Err(format!("Peer unreachable: {}", e))
            }
        }
    }

    /// Handshake: retrieve the peer's verified identity.
    pub fn handshake(&self) -> Result<super::task_board::PeerIdentity, String> {
        let url = format!("{}/peer/identity", self.base_url);
        match self.signed_get(&url) {
            Ok(resp) if resp.status().as_u16() == 200 => {
                self.reachable.store(true, Ordering::Relaxed);
                let text = resp
                    .into_body()
                    .read_to_string()
                    .map_err(|e| e.to_string())?;
                serde_json::from_str(&text).map_err(|e| e.to_string())
            }
            Ok(resp) => {
                self.reachable.store(true, Ordering::Relaxed);
                Err(format!(
                    "Peer handshake failed: HTTP {}",
                    resp.status().as_u16()
                ))
            }
            Err(e) => {
                self.reachable.store(false, Ordering::Relaxed);
                Err(format!("Peer handshake failed: {}", e))
            }
        }
    }

    /// Offer a task to this peer for execution on its local workers.
    /// Returns the result if the peer accepts, or an error if rejected / unreachable.
    pub fn offer_task(
        &self,
        offer: &super::task_board::TaskOffer,
    ) -> Result<super::task_board::TaskResult, String> {
        let url = format!("{}/peer/tasks/offer", self.base_url);
        let body = serde_json::to_vec(offer).map_err(|e| e.to_string())?;

        let timeout = if offer.timeout_secs > 0.0 {
            offer.timeout_secs + 10.0
        } else {
            310.0
        };
        // Reuse the pooled `self.agent` (keep-alive) instead of minting a fresh
        // Agent per dispatch — a new Agent has an empty connection pool and pays
        // a full TCP+TLS handshake RTT on every job. We only need a per-offer
        // deadline, which ureq 3.x lets us override at request scope while still
        // drawing the connection from the shared pool.
        let mut rb = self
            .agent
            .post(&url)
            .config()
            .timeout_global(Some(Duration::from_secs_f64(timeout)))
            .build()
            .header("X-Peer-Key", &self.peer_key)
            .header("Content-Type", "application/json");
        for (k, v) in self.node_headers("POST", "/peer/tasks/offer", &body) {
            rb = rb.header(k.as_str(), v.as_str());
        }
        match rb.send(&body) {
            Ok(resp) if resp.status().as_u16() == 200 => {
                self.reachable.store(true, Ordering::Relaxed);
                let text = resp
                    .into_body()
                    .read_to_string()
                    .map_err(|e| e.to_string())?;
                serde_json::from_str(&text).map_err(|e| e.to_string())
            }
            Ok(resp) if resp.status().as_u16() == 404 => {
                self.reachable.store(true, Ordering::Relaxed);
                let text = resp.into_body().read_to_string().unwrap_or_default();
                Err(format!("Peer rejected task: {}", text))
            }
            Ok(resp) => {
                self.reachable.store(true, Ordering::Relaxed);
                let code = resp.status().as_u16();
                let text = resp.into_body().read_to_string().unwrap_or_default();
                Err(format!("Peer error ({}): {}", code, text))
            }
            Err(e) => {
                self.reachable.store(false, Ordering::Relaxed);
                Err(format!("Peer unreachable: {}", e))
            }
        }
    }

    /// Query user's cached balance on the authoritative peer.
    pub fn get_balance(&self, user_id: &str) -> Result<f64, String> {
        let url = format!("{}/peer/balance?user_id={}", self.base_url, user_id);

        match self.signed_get(&url) {
            Ok(resp) if resp.status().as_u16() == 200 => {
                self.reachable.store(true, Ordering::Relaxed);
                let text = resp
                    .into_body()
                    .read_to_string()
                    .map_err(|e| e.to_string())?;
                let parsed: PeerBalanceResponse =
                    serde_json::from_str(&text).map_err(|e| e.to_string())?;
                Ok(parsed.balance)
            }
            Ok(resp) => {
                self.reachable.store(true, Ordering::Relaxed);
                Err(format!("Peer returned HTTP {}", resp.status().as_u16()))
            }
            Err(e) => {
                self.reachable.store(false, Ordering::Relaxed);
                Err(format!("Peer unreachable: {}", e))
            }
        }
    }

    // --- Async variants (reqwest) for the inline /execute fast path ---
    //
    // The sync methods above run on the broker's bounded `spawn_blocking` pool.
    // These mirror them exactly on the broker's shared async reqwest client so
    // the P2P credit ops can run directly on the tokio runtime. That is the
    // prerequisite for extending the inline `/execute` fast path to the P2P mesh:
    // today the auto-enable predicate keeps inline OFF whenever P2P is enabled,
    // because a synchronous peer call on an async worker thread would stall it.
    // Same wire contract as the sync versions. NOT yet wired into the hot path —
    // the routing/authority restructure + mesh validation land in a follow-up.

    /// Async variant of [`reserve`](Self::reserve).
    #[allow(dead_code)]
    pub async fn reserve_async(
        &self,
        http: &reqwest::Client,
        user_id: &str,
        amount: f64,
        request_id: &str,
    ) -> Result<PeerReserveResponse, String> {
        let url = format!("{}/peer/reserve", self.base_url);
        let req = PeerReserveRequest {
            user_id: user_id.to_string(),
            amount,
            request_id: request_id.to_string(),
        };
        let body = serde_json::to_vec(&req).map_err(|e| e.to_string())?;
        match self.signed_reqwest_post(http, &url, body).send().await {
            Ok(resp) => {
                self.reachable.store(true, Ordering::Relaxed);
                let code = resp.status().as_u16();
                let text = resp.text().await.map_err(|e| e.to_string())?;
                if code == 200 {
                    serde_json::from_str(&text).map_err(|e| e.to_string())
                } else if let Ok(err) = serde_json::from_str::<PeerErrorResponse>(&text) {
                    Err(format!("[{}] {}", code, err.error))
                } else {
                    Err(format!("Peer returned {}: {}", code, text))
                }
            }
            Err(e) => {
                self.reachable.store(false, Ordering::Relaxed);
                Err(format!("Peer unreachable: {}", e))
            }
        }
    }

    /// Async variant of [`commit`](Self::commit).
    #[allow(dead_code)]
    pub async fn commit_async(
        &self,
        http: &reqwest::Client,
        reservation_id: &str,
        actual_cost: f64,
    ) -> Result<PeerCommitResponse, String> {
        let url = format!("{}/peer/commit", self.base_url);
        let req = PeerCommitRequest {
            reservation_id: reservation_id.to_string(),
            actual_cost,
        };
        let body = serde_json::to_vec(&req).map_err(|e| e.to_string())?;
        match self.signed_reqwest_post(http, &url, body).send().await {
            Ok(resp) => {
                self.reachable.store(true, Ordering::Relaxed);
                let code = resp.status().as_u16();
                let text = resp.text().await.map_err(|e| e.to_string())?;
                if code == 200 {
                    serde_json::from_str(&text).map_err(|e| e.to_string())
                } else {
                    Err(format!("Peer commit failed ({}): {}", code, text))
                }
            }
            Err(e) => {
                self.reachable.store(false, Ordering::Relaxed);
                Err(format!("Peer unreachable: {}", e))
            }
        }
    }

    /// Async variant of [`cancel`](Self::cancel). A non-2xx means the cancel was
    /// rejected, not applied (audit M2) — same semantics as the sync version.
    #[allow(dead_code)]
    pub async fn cancel_async(
        &self,
        http: &reqwest::Client,
        reservation_id: &str,
    ) -> Result<(), String> {
        let url = format!("{}/peer/cancel", self.base_url);
        let req = PeerCancelRequest {
            reservation_id: reservation_id.to_string(),
        };
        let body = serde_json::to_vec(&req).map_err(|e| e.to_string())?;
        match self.signed_reqwest_post(http, &url, body).send().await {
            Ok(resp) => {
                self.reachable.store(true, Ordering::Relaxed);
                let code = resp.status().as_u16();
                if (200..300).contains(&code) {
                    Ok(())
                } else {
                    Err(format!("Peer cancel failed: HTTP {}", code))
                }
            }
            Err(e) => {
                self.reachable.store(false, Ordering::Relaxed);
                Err(format!("Peer unreachable: {}", e))
            }
        }
    }

    /// Async variant of [`get_balance`](Self::get_balance).
    #[allow(dead_code)]
    pub async fn get_balance_async(
        &self,
        http: &reqwest::Client,
        user_id: &str,
    ) -> Result<f64, String> {
        let url = format!("{}/peer/balance?user_id={}", self.base_url, user_id);
        match self.signed_reqwest_get(http, &url).send().await {
            Ok(resp) if resp.status().as_u16() == 200 => {
                self.reachable.store(true, Ordering::Relaxed);
                let text = resp.text().await.map_err(|e| e.to_string())?;
                let parsed: PeerBalanceResponse =
                    serde_json::from_str(&text).map_err(|e| e.to_string())?;
                Ok(parsed.balance)
            }
            Ok(resp) => {
                self.reachable.store(true, Ordering::Relaxed);
                Err(format!("Peer returned HTTP {}", resp.status().as_u16()))
            }
            Err(e) => {
                self.reachable.store(false, Ordering::Relaxed);
                Err(format!("Peer unreachable: {}", e))
            }
        }
    }
}

impl std::fmt::Debug for PeerClient {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PeerClient")
            .field("base_url", &self.base_url)
            .field("reachable", &self.is_reachable())
            .finish()
    }
}

// --- PeerManager ---

/// Manages peer connections and authority assignment.
/// Extract the host of a peer base URL: "http://10.0.0.1:9000" -> "10.0.0.1".
fn host_of(url: &str) -> &str {
    let no_scheme = url
        .strip_prefix("http://")
        .or_else(|| url.strip_prefix("https://"))
        .unwrap_or(url);
    let authority = no_scheme.split('/').next().unwrap_or("");
    authority
        .rsplit_once(':')
        .map(|(h, _)| h)
        .unwrap_or(authority)
}

/// Extract the port of a peer base URL, defaulting by scheme when absent.
fn port_of(url: &str) -> u16 {
    let (default, no_scheme) = match url.strip_prefix("http://") {
        Some(rest) => (80u16, rest),
        None => match url.strip_prefix("https://") {
            Some(rest) => (443u16, rest),
            None => (80u16, url),
        },
    };
    no_scheme
        .split('/')
        .next()
        .unwrap_or("")
        .rsplit_once(':')
        .and_then(|(_, p)| p.parse().ok())
        .unwrap_or(default)
}

/// Every socket address a peer base URL's authority resolves to.
///
/// Returns an empty set rather than an error when the host does not resolve:
/// callers use set intersection, and an empty set intersects nothing, so an
/// unresolvable peer can never be matched by accident.
fn endpoint_addrs(url: &str) -> std::collections::HashSet<SocketAddr> {
    (host_of(url), port_of(url))
        .to_socket_addrs()
        .map(|it| it.collect())
        .unwrap_or_default()
}

pub struct PeerManager {
    /// peer base URL → client
    peers: DashMap<String, PeerClient>,
    /// This broker's index in the sorted peer list (0-based)
    own_broker_index: u8,
    /// Total number of brokers (including self)
    total_brokers: u8,
    /// Shared secret for peer auth
    peer_key: String,
    /// This node's signing identity, handed to every PeerClient for X-Node-* signing.
    node_key: Option<Arc<NodeKey>>,
    /// P2P mode enabled
    enabled: bool,
    /// Cached QUIC ports per peer URL (populated from /peer/identity)
    quic_ports: DashMap<String, u16>,
    /// Base URLs of peers learned from the dashboard roster, as opposed to
    /// those configured via `ZAKURO_PEERS`. Membership here is the ONLY thing
    /// that makes a peer evictable — see [`remove_roster_peer`](Self::remove_roster_peer).
    /// A configured peer is never inserted, so the cluster-native fallback
    /// path cannot be reconciled away no matter what the roster says.
    roster_derived: DashMap<String, ()>,
}

impl PeerManager {
    /// Create a new PeerManager.
    ///
    /// `own_ip` — this broker's WireGuard IP (or hostname).
    /// `peer_addresses` — list of peer broker "ip:port" strings (from ZAKURO_PEERS).
    /// `broker_port` — the port this broker listens on (default 9000).
    /// `peer_key` — shared secret from ZAKURO_PEER_KEY env var.
    /// `enabled` — whether P2P is enabled (ZAKURO_P2P=true).
    pub fn new(
        own_ip: Option<&str>,
        peer_addresses: &[String],
        broker_port: u16,
        peer_key: String,
        enabled: bool,
    ) -> Self {
        Self::new_with_node_key(own_ip, peer_addresses, broker_port, peer_key, enabled, None)
    }

    /// Like [`new`](Self::new) but with this node's signing key for outbound
    /// `/peer/*` request signatures.
    #[allow(clippy::too_many_arguments)]
    pub fn new_with_node_key(
        own_ip: Option<&str>,
        peer_addresses: &[String],
        broker_port: u16,
        peer_key: String,
        enabled: bool,
        node_key: Option<Arc<NodeKey>>,
    ) -> Self {
        let peers = DashMap::new();

        // Build a sorted set of all broker addresses (self + peers) for deterministic hashing.
        let mut all_addresses = BTreeSet::new();

        // Add self
        let own_addr = if let Some(ip) = own_ip {
            format!("{}:{}", ip, broker_port)
        } else {
            format!("127.0.0.1:{}", broker_port)
        };
        all_addresses.insert(own_addr.clone());

        // Add peers — preserve original port for HTTP calls,
        // but use normalized addr (ip:broker_port) for the hash ring.
        let mut peer_url_map: std::collections::HashMap<String, String> =
            std::collections::HashMap::new();
        for peer in peer_addresses {
            if peer.is_empty() {
                continue;
            }
            let raw = peer
                .strip_prefix("http://")
                .or_else(|| peer.strip_prefix("https://"))
                .unwrap_or(peer);
            let actual_url = if raw.contains("://") {
                peer.to_string()
            } else {
                format!("http://{}", raw)
            };
            // Normalize for hash ring (consistent ordering)
            let (ip, _) = raw.rsplit_once(':').unwrap_or((raw, ""));
            let norm_addr = format!("{}:{}", ip, broker_port);
            all_addresses.insert(norm_addr.clone());
            peer_url_map.insert(norm_addr, actual_url);
        }

        // Determine own index in sorted order
        let sorted: Vec<String> = all_addresses.into_iter().collect();
        let own_index = sorted.iter().position(|a| *a == own_addr).unwrap_or(0) as u8;
        let total = sorted.len() as u8;

        // Create PeerClients using the ACTUAL peer URLs (not normalized)
        if enabled {
            for addr in &sorted {
                if *addr != own_addr {
                    let base_url = peer_url_map
                        .get(addr)
                        .cloned()
                        .unwrap_or_else(|| format!("http://{}", addr));
                    let client = PeerClient::new_with_node_key(
                        base_url.clone(),
                        peer_key.clone(),
                        node_key.clone(),
                    );
                    eprintln!("  [P2P] Registered peer broker at {}", base_url);
                    peers.insert(base_url, client);
                }
            }
        }

        eprintln!(
            "  [P2P] Broker index={}/{} (p2p={})",
            own_index, total, enabled
        );

        Self {
            peers,
            own_broker_index: own_index,
            total_brokers: total,
            peer_key,
            node_key,
            enabled,
            quic_ports: DashMap::new(),
            roster_derived: DashMap::new(),
        }
    }

    /// Determine which broker is authoritative for a given user_id.
    pub fn determine_authority(&self, user_id: &str) -> Authority {
        if !self.enabled || self.total_brokers <= 1 {
            return Authority::Standalone;
        }

        let hash = simple_hash(user_id);
        let owner_index = (hash % self.total_brokers as u64) as u8;

        if owner_index == self.own_broker_index {
            Authority::Local
        } else {
            // Find the peer URL for this index.
            // IMPORTANT: DashMap iteration is unordered. Sort peer URLs so the mapping
            // from owner_index → URL is deterministic across calls and broker restarts.
            let mut peer_urls: Vec<String> = self.peers.iter().map(|e| e.key().clone()).collect();
            peer_urls.sort();
            // Index 0..N covers all brokers in sorted order; skip own_broker_index.
            // If owner_index < own_broker_index: peer_idx = owner_index
            // If owner_index > own_broker_index: peer_idx = owner_index - 1
            let peer_idx = if owner_index < self.own_broker_index {
                owner_index as usize
            } else {
                (owner_index - 1) as usize
            };

            if peer_idx < peer_urls.len() {
                let url = &peer_urls[peer_idx];
                // Check if peer is reachable
                if let Some(client) = self.peers.get(url) {
                    if client.is_reachable() {
                        return Authority::Peer(url.clone());
                    }
                }
            }
            // Peer unreachable — fall back to standalone (use local in-memory + dashboard API)
            Authority::Standalone
        }
    }

    /// Get a PeerClient by base URL.
    pub fn get_client(
        &self,
        url: &str,
    ) -> Option<dashmap::mapref::one::Ref<'_, String, PeerClient>> {
        self.peers.get(url)
    }

    /// Find the base URL for the peer whose host equals the given IP.
    /// Used to locate the peer broker responsible for a given worker IP.
    /// Matches the URL host EXACTLY — a substring match would let "10.0.0.1"
    /// resolve to "10.0.0.10" and mis-route earn/credit calls (audit M2).
    pub fn get_url_for_ip(&self, ip: &str) -> Option<String> {
        self.peers
            .iter()
            .find(|e| host_of(e.key()) == ip)
            .map(|e| e.key().clone())
    }

    /// Find the registered peer that is the SAME endpoint as `url`, comparing
    /// resolved socket addresses rather than URL strings.
    ///
    /// A peer identifies itself to its publisher by the address it believes it
    /// has -- `http://<own_ip>:<port>` (see `our_url` in `start_server`) -- but
    /// it is REGISTERED under whatever `ZAKURO_PEERS` spelled, which may be a
    /// DNS name. Those two spellings are the same host, so an exact-string
    /// lookup misses and settlement lands in the pay-nobody arm: the executor
    /// is never paid and the requester is fully refunded, silently, for real
    /// work that really ran.
    ///
    /// IP-configured deployments never saw this because the two spellings
    /// coincide there. The exact-match case is checked first and costs nothing,
    /// so those deployments keep their current behaviour and take no lookup.
    ///
    /// Matching requires an identical resolved `SocketAddr` -- host AND port --
    /// so this cannot do what the `get_url_for_ip` audit warned about and let
    /// "10.0.0.1" match "10.0.0.10", nor pay a peer that merely shares a host
    /// on a different port.
    pub fn get_url_for_endpoint(&self, url: &str) -> Option<String> {
        if self.peers.contains_key(url) {
            return Some(url.to_string());
        }
        let want = endpoint_addrs(url);
        if want.is_empty() {
            return None;
        }
        self.peers
            .iter()
            .find(|e| !endpoint_addrs(e.key()).is_disjoint(&want))
            .map(|e| e.key().clone())
    }

    /// Find the base URL for the peer whose cached key-derived identity has this
    /// fingerprint. Identity-keyed (not address-keyed), so it resolves correctly
    /// for a peer reached via a proxy/NAT whose base-URL host is not its worker's
    /// IP — the case `get_url_for_ip` silently misses. An empty `fp` never matches.
    ///
    /// TOTAL over both identity forms: the query is normalized with
    /// `strip_node_arg` just like the stored value, so `"<fp>"` and the canonical
    /// `"zc://node-<fp>"` resolve identically. Previously only the STORED side was
    /// stripped, so a caller passing the canonical form — the form
    /// `TaskResult::executor_node_uri` and `NodeIdentity::node_uri()` both produce —
    /// got a silent `None` that routed settlement into the pay-nobody arm. This
    /// accepts strictly more than before and never less, so no caller can break.
    pub fn get_url_for_fingerprint(&self, fp: &str) -> Option<String> {
        let fp = super::node_identity::strip_node_arg(fp);
        if fp.is_empty() {
            return None;
        }
        self.peers
            .iter()
            .find(|e| {
                e.value()
                    .fingerprint()
                    .map(|f| super::node_identity::strip_node_arg(&f) == fp)
                    .unwrap_or(false)
            })
            .map(|e| e.key().clone())
    }

    /// Test/bootstrap seam: record a peer's key-derived identity as if a signed
    /// `/peer/health` probe had just reported it (see `PeerClient::check_health`).
    pub fn set_peer_fingerprint(&self, peer_url: &str, fp: &str) {
        if let Some(c) = self.peers.get(peer_url) {
            c.set_fingerprint(fp);
        }
    }

    /// Register or update a peer broker.
    pub fn register_peer(&self, base_url: String) {
        if !self.peers.contains_key(&base_url) {
            let client = PeerClient::new_with_node_key(
                base_url.clone(),
                self.peer_key.clone(),
                self.node_key.clone(),
            );
            self.peers.insert(base_url, client);
        }
    }

    /// Register a peer learned from the dashboard roster, marking it
    /// evictable so [`retain_roster_peers`](Self::retain_roster_peers) can
    /// drop it once it leaves the roster.
    ///
    /// Registering a URL that is ALREADY known as a configured
    /// (`ZAKURO_PEERS`) peer does not mark it roster-derived: provenance only
    /// ever moves toward "permanent", never away from it, so the roster can
    /// never make the fallback path evictable.
    pub fn register_roster_peer(&self, base_url: String) {
        let already_configured =
            self.peers.contains_key(&base_url) && !self.roster_derived.contains_key(&base_url);
        self.register_peer(base_url.clone());
        if !already_configured {
            self.roster_derived.insert(base_url, ());
        }
    }

    /// Whether the peer at `base_url` was learned from the roster (and is
    /// therefore evictable). Configured peers and unknown URLs are `false`.
    pub fn is_roster_derived(&self, base_url: &str) -> bool {
        self.roster_derived.contains_key(base_url)
    }

    /// Base URLs of every roster-derived peer currently registered.
    pub fn roster_derived_urls(&self) -> Vec<String> {
        self.roster_derived
            .iter()
            .map(|e| e.key().clone())
            .collect()
    }

    /// Drop a roster-derived peer and every index keyed on its URL.
    ///
    /// Returns `false` — changing nothing — for a configured (`ZAKURO_PEERS`)
    /// peer or an unknown URL. That guard lives HERE, in the single removal
    /// path, rather than at each call site, so no future caller can evict the
    /// cluster-native fallback peers by mistake.
    pub fn remove_roster_peer(&self, base_url: &str) -> bool {
        if !self.roster_derived.contains_key(base_url) {
            return false;
        }
        self.peers.remove(base_url);
        self.quic_ports.remove(base_url);
        self.roster_derived.remove(base_url);
        true
    }

    /// Evict every roster-derived peer whose URL is not in `keep`.
    ///
    /// Configured peers are untouched regardless of `keep` (see
    /// [`remove_roster_peer`](Self::remove_roster_peer)), so passing an empty
    /// `keep` — what an endpoint-less roster produces — is a no-op for a
    /// broker running purely on `ZAKURO_PEERS`.
    ///
    /// Returns the URLs actually evicted.
    pub fn retain_roster_peers(&self, keep: &std::collections::HashSet<String>) -> Vec<String> {
        let mut evicted = Vec::new();
        for url in self.roster_derived_urls() {
            if !keep.contains(&url) && self.remove_roster_peer(&url) {
                evicted.push(url);
            }
        }
        evicted
    }

    /// The base URL of the CONFIGURED (`ZAKURO_PEERS`) peer whose cached
    /// identity has this fingerprint, if any.
    ///
    /// This is the dedup primitive for roster merging: one physical broker
    /// appears as `http://zc-broker-1.zc-broker...:9000` from `ZAKURO_PEERS`
    /// and as `http://10.13.13.7:9000` from the roster, and only the
    /// fingerprint can tell that those two spellings are one broker. It is
    /// deliberately restricted to configured peers so the answer is stable:
    /// resolving against *any* spelling would return whichever entry
    /// `DashMap` iteration happened to reach first once a duplicate existed,
    /// and the caller could never converge on removing it.
    ///
    /// `None` also means "not resolved yet" — a peer that has never been
    /// successfully probed has no fingerprint. Callers must treat that as
    /// "unknown", never as "different broker" (see the tick loop in
    /// `server.rs`).
    pub fn configured_url_for_fingerprint(&self, fp: &str) -> Option<String> {
        let fp = super::node_identity::strip_node_arg(fp);
        if fp.is_empty() {
            return None;
        }
        self.peers
            .iter()
            .find(|e| {
                !self.roster_derived.contains_key(e.key())
                    && e.value()
                        .fingerprint()
                        .map(|f| super::node_identity::strip_node_arg(&f) == fp)
                        .unwrap_or(false)
            })
            .map(|e| e.key().clone())
    }

    /// Check health of all peers and update reachability.
    pub fn health_check_all(&self) {
        for entry in self.peers.iter() {
            entry.value().check_health();
        }
    }

    /// Whether P2P is enabled.
    pub fn is_enabled(&self) -> bool {
        self.enabled
    }

    /// Get peer key for verifying incoming requests.
    pub fn peer_key(&self) -> &str {
        &self.peer_key
    }

    /// Number of peers (excluding self).
    pub fn peer_count(&self) -> usize {
        self.peers.len()
    }

    /// List all peer URLs.
    pub fn peer_urls(&self) -> Vec<String> {
        self.peers.iter().map(|e| e.key().clone()).collect()
    }

    /// The cached key-derived identity (`zc://node-<fp>`) for a peer, if
    /// learned yet (via a prior successful `/peer/health` probe). IP-free —
    /// this is the only way `/brokers` should resolve a peer's id; it must
    /// never fall back to exposing `peer_url` (which carries the IP).
    pub fn peer_fingerprint(&self, peer_url: &str) -> Option<String> {
        self.peers.get(peer_url).and_then(|c| c.fingerprint())
    }

    /// Cache a freshly fetched-and-verified advertisement for the peer at `url`.
    /// No-op if `url` isn't a known peer.
    pub fn set_peer_advert(&self, url: &str, advert: super::discovery::Advert) {
        if let Some(client) = self.peers.get(url) {
            client.set_advert(advert);
        }
    }

    /// The most recently cached advertisement for the peer at `url`, if any.
    pub fn peer_advert(&self, url: &str) -> Option<super::discovery::Advert> {
        self.peers.get(url).and_then(|c| c.advert())
    }

    /// Round-trip time (ms) of the peer at `url`'s most recent successful
    /// `/peer/health` probe (see `PeerClient::check_health`), or `None` if
    /// unknown/never probed. Used as the ranking signal for marketplace
    /// pre-selection (task 5) — "lowest observed latency" policy.
    pub fn peer_latency_ms(&self, url: &str) -> Option<f64> {
        self.peers.get(url).and_then(|c| c.last_rtt_ms())
    }

    /// Best-effort: ensure every peer's fingerprint is cached, probing any
    /// peer that hasn't been health-checked yet (or whose prior probe didn't
    /// yield a fingerprint). Used by `/brokers` so a cold-started broker
    /// still returns a useful listing on first request rather than only
    /// after the next periodic health-check tick.
    pub fn ensure_fingerprints_cached(&self) {
        for entry in self.peers.iter() {
            if entry.value().fingerprint().is_none() {
                entry.value().check_health();
            }
        }
    }

    /// Store the QUIC port for a given peer URL.
    pub fn set_quic_port(&self, peer_url: &str, port: u16) {
        if port > 0 {
            self.quic_ports.insert(peer_url.to_string(), port);
        }
    }

    /// Get the cached QUIC port for a peer URL (0 = unknown).
    pub fn get_quic_port(&self, peer_url: &str) -> u16 {
        self.quic_ports.get(peer_url).map(|v| *v).unwrap_or(0)
    }

    /// Best-effort: write the current peer set to `<dir>/peers.json`.
    /// Never panics — filesystem errors are silently ignored.
    pub fn persist_to(&self, dir: &std::path::Path) {
        let entries: Vec<PersistedPeer> = self
            .peers
            .iter()
            .map(|e| {
                let url = e.key().clone();
                let fp = self.peer_fingerprint(&url).unwrap_or_default();
                let last_seen = std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .map(|d| d.as_secs())
                    .unwrap_or(0);
                let roster_derived = self.roster_derived.contains_key(&url);
                PersistedPeer {
                    fp,
                    url,
                    last_seen,
                    roster_derived,
                }
            })
            .collect();

        if std::fs::create_dir_all(dir).is_err() {
            return;
        }
        if let Ok(json) = serde_json::to_string(&entries) {
            let _ = std::fs::write(dir.join("peers.json"), json);
        }
    }

    /// Best-effort: register peers cached at `<dir>/peers.json` (from a
    /// previous run) so they're known immediately on startup, as unhealthy
    /// hints (a subsequent health check confirms reachability). Never
    /// panics — filesystem/parse errors are silently ignored.
    pub fn load_persisted_from(&self, dir: &std::path::Path) {
        let Ok(text) = std::fs::read_to_string(dir.join("peers.json")) else {
            return;
        };
        let Ok(entries) = serde_json::from_str::<Vec<PersistedPeer>>(&text) else {
            return;
        };
        for entry in entries {
            let url = entry.url;
            if entry.roster_derived {
                self.register_roster_peer(url.clone());
            } else {
                self.register_peer(url.clone());
            }
            // Seed the identity cache from the fingerprint THIS node persisted
            // (written by `persist_to` from an already health-verified
            // `PeerClient::fingerprint()`). Without this, every restart leaves
            // the cache empty until the next `health_check_all` tick (~60s),
            // during which `get_url_for_fingerprint` resolves nothing and the
            // earn path pays nobody for real remote work — peers' *workers* are
            // stamped independently from the sync payload's `source_node`, so
            // they stay routable and executable throughout that window.
            if !entry.fp.is_empty() {
                self.set_peer_fingerprint(&url, &entry.fp);
            }
        }
    }

    /// Write the current peer set to `~/.zakuro/peers.json` (best-effort;
    /// skipped silently when the credentials dir is unavailable).
    pub fn persist(&self) {
        if let Some(dir) = crate::credentials::dir() {
            self.persist_to(&dir);
        }
    }

    /// Load the cached peer set from `~/.zakuro/peers.json` (best-effort;
    /// skipped silently when the credentials dir is unavailable).
    pub fn load_persisted(&self) {
        if let Some(dir) = crate::credentials::dir() {
            self.load_persisted_from(&dir);
        }
    }

    /// Snapshot the current peer set as `{fp, url, epoch}` gossip entries for
    /// `GET /peer/peers`. `epoch` is a freshness stamp (current time at
    /// snapshot) — the same convention `persist_to` uses for
    /// `PersistedPeer::last_seen`. It is a forward-compatibility hook for a
    /// future last-writer-wins/delta merge (spec #2 §6.5); this task always
    /// gossips the full peer set.
    pub fn gossip_entries(&self) -> Vec<PersistedPeer> {
        let last_seen = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);
        self.peers
            .iter()
            .map(|e| {
                let url = e.key().clone();
                let fp = self.peer_fingerprint(&url).unwrap_or_default();
                let roster_derived = self.roster_derived.contains_key(&url);
                PersistedPeer {
                    fp,
                    url,
                    last_seen,
                    roster_derived,
                }
            })
            .collect()
    }

    /// Fetch and cache QUIC ports from all peers via /peer/identity.
    pub fn discover_quic_ports(&self) {
        for entry in self.peers.iter() {
            if let Ok(identity) = entry.value().handshake() {
                if identity.quic_port > 0 {
                    self.quic_ports
                        .insert(entry.key().clone(), identity.quic_port);
                    eprintln!(
                        "  [QUIC] Discovered peer {} QUIC port: {}",
                        entry.key(),
                        identity.quic_port
                    );
                }
            }
        }
    }
}

impl std::fmt::Debug for PeerManager {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PeerManager")
            .field("own_broker_index", &self.own_broker_index)
            .field("total_brokers", &self.total_brokers)
            .field("enabled", &self.enabled)
            .field("peer_count", &self.peers.len())
            .finish()
    }
}

/// Verify an incoming peer request's X-Peer-Key value against the configured
/// key. `provided` is the value of the request's `X-Peer-Key` header (or `None`
/// when absent). A keyless broker (`expected_key` empty) rejects all peers.
pub fn verify_peer_key(provided: Option<&str>, expected_key: &str) -> bool {
    if expected_key.is_empty() {
        return false; // No key configured — reject all peer requests
    }
    provided.map(|k| k == expected_key).unwrap_or(false)
}

/// Simple deterministic hash for user_id → broker assignment.
/// Uses FNV-1a for speed and good distribution.
pub fn simple_hash(s: &str) -> u64 {
    let mut hash: u64 = 0xcbf29ce484222325; // FNV offset basis
    for byte in s.as_bytes() {
        hash ^= *byte as u64;
        hash = hash.wrapping_mul(0x100000001b3); // FNV prime
    }
    hash
}

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

    #[test]
    fn persist_then_load_roundtrips_peer_urls() {
        let dir = std::env::temp_dir().join(format!(
            "zc2-peer-persist-test-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let pm = PeerManager::new(Some("10.13.13.9"), &[], 9000, "k".into(), true);
        pm.register_peer("http://10.13.13.17:9000".into());
        pm.persist_to(&dir);
        let pm2 = PeerManager::new(Some("10.13.13.9"), &[], 9000, "k".into(), true);
        pm2.load_persisted_from(&dir);
        assert!(pm2.peer_urls().iter().any(|u| u.contains("10.13.13.17")));
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn host_of_extracts_host() {
        assert_eq!(host_of("http://10.0.0.1:9000"), "10.0.0.1");
        assert_eq!(host_of("https://example.com:443/x"), "example.com");
        assert_eq!(host_of("10.0.0.5:3960"), "10.0.0.5");
    }

    #[test]
    fn url_sign_path_strips_host_and_query() {
        assert_eq!(
            url_sign_path("http://10.13.13.5:9000/peer/reserve"),
            "/peer/reserve"
        );
        assert_eq!(
            url_sign_path("http://h:9000/peer/balance?user_id=1"),
            "/peer/balance"
        );
    }

    #[test]
    fn outbound_signature_verifies_server_side() {
        use crate::broker::node_identity::{verify_request, ReplayGuard};
        let key = Arc::new(NodeKey::generate());
        let client = PeerClient::new_with_node_key(
            "http://10.13.13.5:9000".into(),
            "peerkey".into(),
            Some(key.clone()),
        );
        let body = br#"{"user_id":"u1","amount":2.0,"request_id":"r1"}"#;
        let hdrs = client.node_headers("POST", "/peer/reserve", body);
        // the server would look these up as request headers:
        let get = |name: &str| -> Option<String> {
            hdrs.iter().find(|(h, _)| h == name).map(|(_, v)| v.clone())
        };
        let signer = key.public_b64();
        let rostered = |id: &str| id == signer;
        let guard = ReplayGuard::new();
        let now = crate::broker::node_identity::now_secs();
        let got = verify_request(&rostered, &guard, "POST", "/peer/reserve", body, &get, now)
            .expect("server accepts the signed request");
        assert_eq!(got, signer);
    }

    #[test]
    fn no_node_key_produces_no_headers() {
        let client = PeerClient::new("http://h:9000".into(), "k".into());
        assert!(client
            .node_headers("POST", "/peer/reserve", b"{}")
            .is_empty());
    }

    #[test]
    fn get_url_for_ip_matches_host_exactly_not_substring() {
        // own_ip distinct from both peers so neither peer is dropped as "self".
        // Two peers whose IPs are substrings of each other.
        let pm = PeerManager::new(
            Some("10.0.0.99"),
            &[
                "http://10.0.0.10:9000".to_string(),
                "http://10.0.0.1:9000".to_string(),
            ],
            9000,
            "k".to_string(),
            true,
        );
        // "10.0.0.1" must resolve to the .1 peer, never the .10 peer.
        let got = pm.get_url_for_ip("10.0.0.1");
        assert_eq!(
            got.as_deref(),
            Some("http://10.0.0.1:9000"),
            "exact host match required; got {:?}",
            got
        );
        // "10.0.0.10" resolves to its own peer.
        assert_eq!(
            pm.get_url_for_ip("10.0.0.10").as_deref(),
            Some("http://10.0.0.10:9000")
        );
        // An IP that is a substring of a peer but not an exact host → None.
        assert_eq!(pm.get_url_for_ip("0.0.0.1"), None);
    }

    #[test]
    fn get_url_for_fingerprint_matches_cached_identity_not_host() {
        let pm = PeerManager::new(
            Some("10.13.13.2"),
            &["10.13.13.7:9000".to_string(), "10.13.13.8:9000".to_string()],
            9000,
            "k".to_string(),
            true,
        );
        // Simulate what a successful signed /peer/health probe caches.
        pm.set_peer_fingerprint("http://10.13.13.7:9000", "aaaaaaaaaaaaaaaa");
        pm.set_peer_fingerprint("http://10.13.13.8:9000", "bbbbbbbbbbbbbbbb");

        assert_eq!(
            pm.get_url_for_fingerprint("bbbbbbbbbbbbbbbb").as_deref(),
            Some("http://10.13.13.8:9000")
        );
        // The function is TOTAL over both identity forms: the canonical
        // `zc://node-<fp>` query resolves to the same peer as bare hex. This is
        // the form `TaskResult::executor_node_uri` carries, so a future consumer
        // passing it straight through must NOT silently fall into the
        // pay-nobody settlement arm.
        assert_eq!(
            pm.get_url_for_fingerprint("zc://node-bbbbbbbbbbbbbbbb")
                .as_deref(),
            Some("http://10.13.13.8:9000")
        );
        // The intermediate `node-<fp>` form too, for symmetry with strip_node_arg.
        assert_eq!(
            pm.get_url_for_fingerprint("node-bbbbbbbbbbbbbbbb")
                .as_deref(),
            Some("http://10.13.13.8:9000")
        );

        // Unknown fingerprint resolves to nothing — never a lucky partial match.
        assert_eq!(pm.get_url_for_fingerprint("bbbbbbbbbbbbbbb"), None);
        assert_eq!(pm.get_url_for_fingerprint(""), None);
        // Accepting the prefixed form must not make the prefix ALONE match.
        assert_eq!(pm.get_url_for_fingerprint("zc://node-"), None);
        // Case must match exactly — never case-insensitive.
        assert_eq!(pm.get_url_for_fingerprint("BBBBBBBBBBBBBBBB"), None);
        assert_eq!(
            pm.get_url_for_fingerprint("zc://node-BBBBBBBBBBBBBBBB"),
            None
        );
    }

    #[test]
    fn proxied_peer_resolves_by_identity_where_ip_match_fails() {
        // The reporting user's Mac: `access: proxy 127.0.0.1:18888`. The peer is
        // registered at its proxied base URL; its worker advertises a mesh IP
        // ("10.13.13.7") that never equals the base-URL host ("127.0.0.1").
        let pm = PeerManager::new(
            Some("10.13.13.2"),
            &["127.0.0.1:18888".to_string()],
            9000,
            "k".to_string(),
            true,
        );
        // Simulate what a successful signed /peer/health probe caches: the
        // canonicalized `zc://node-<fp>` form (see PeerClient::set_fingerprint).
        pm.set_peer_fingerprint("http://127.0.0.1:18888", "zc://node-cccccccccccccccc");

        // Old behaviour: the worker's IP never equals the peer's base-URL host,
        // so settlement fell into the refund arm and the executor earned nothing.
        assert_eq!(pm.get_url_for_ip("10.13.13.7"), None);
        // New behaviour: identity resolves regardless of transport, and
        // regardless of which identity form the caller holds — strip_node_arg is
        // now applied to the query as well as the stored value.
        assert_eq!(
            pm.get_url_for_fingerprint("cccccccccccccccc").as_deref(),
            Some("http://127.0.0.1:18888")
        );
        assert_eq!(
            pm.get_url_for_fingerprint("zc://node-cccccccccccccccc")
                .as_deref(),
            Some("http://127.0.0.1:18888")
        );
    }

    #[test]
    fn test_simple_hash_deterministic() {
        let h1 = simple_hash("9000000001");
        let h2 = simple_hash("9000000001");
        assert_eq!(h1, h2);
    }

    #[test]
    fn test_simple_hash_distribution() {
        // With 2 brokers, hashing 100 user IDs should give roughly 50/50 split
        let mut counts = [0u32; 2];
        for i in 9000000001..9000000101u64 {
            let h = simple_hash(&i.to_string());
            counts[(h % 2) as usize] += 1;
        }
        // Allow some skew but both should have at least 20
        assert!(counts[0] >= 20, "Broker 0 got {} users", counts[0]);
        assert!(counts[1] >= 20, "Broker 1 got {} users", counts[1]);
    }

    /// The endpoint lookup is what stands between real remote work and the
    /// pay-nobody arm of `settle_executor_payout`, so each case is pinned.
    ///
    /// "localhost" is used as the DNS name throughout: it is the only name
    /// guaranteed to resolve with no network access.
    #[test]
    fn get_url_for_endpoint_matches_exact_key_first() {
        let pm = PeerManager::new(Some("10.0.0.1"), &[], 9000, "key".into(), true);
        pm.register_peer("http://10.0.0.2:9000".to_string());
        assert_eq!(
            pm.get_url_for_endpoint("http://10.0.0.2:9000").as_deref(),
            Some("http://10.0.0.2:9000"),
            "an exactly-matching key must resolve to itself and take no lookup"
        );
    }

    #[test]
    fn get_url_for_endpoint_matches_ip_against_dns_registration() {
        // The regression: the peer is REGISTERED by name but identifies itself
        // to its publisher by IP. Before this, settlement string-compared the
        // two, missed, and refunded the requester for work that really ran.
        let pm = PeerManager::new(Some("10.0.0.1"), &[], 9000, "key".into(), true);
        pm.register_peer("http://localhost:9000".to_string());
        assert_eq!(
            pm.get_url_for_endpoint("http://127.0.0.1:9000").as_deref(),
            Some("http://localhost:9000"),
            "an IP self-identification must find its DNS-registered peer"
        );
    }

    #[test]
    fn get_url_for_endpoint_matches_dns_against_ip_registration() {
        // The same mismatch in the other direction.
        let pm = PeerManager::new(Some("10.0.0.1"), &[], 9000, "key".into(), true);
        pm.register_peer("http://127.0.0.1:9000".to_string());
        assert_eq!(
            pm.get_url_for_endpoint("http://localhost:9000").as_deref(),
            Some("http://127.0.0.1:9000")
        );
    }

    #[test]
    fn get_url_for_endpoint_requires_the_same_port() {
        // Host alone is not identity: paying a peer listening on another port
        // would credit the wrong broker on a shared host.
        let pm = PeerManager::new(Some("10.0.0.1"), &[], 9000, "key".into(), true);
        pm.register_peer("http://localhost:9000".to_string());
        assert_eq!(pm.get_url_for_endpoint("http://127.0.0.1:9100"), None);
    }

    #[test]
    fn get_url_for_endpoint_does_not_prefix_match_hosts() {
        // The hazard audit M2 called out for `get_url_for_ip`: "10.0.0.1" must
        // never resolve to "10.0.0.10". Comparing resolved SocketAddrs makes
        // this structural rather than a matter of careful string handling.
        let pm = PeerManager::new(Some("10.0.0.1"), &[], 9000, "key".into(), true);
        pm.register_peer("http://10.0.0.10:9000".to_string());
        assert_eq!(pm.get_url_for_endpoint("http://10.0.0.1:9000"), None);
    }

    #[test]
    fn get_url_for_endpoint_returns_none_for_unresolvable_and_unknown() {
        let pm = PeerManager::new(Some("10.0.0.1"), &[], 9000, "key".into(), true);
        pm.register_peer("http://10.0.0.2:9000".to_string());
        // .invalid is reserved by RFC 2606 and never resolves: an empty address
        // set must match nothing rather than matching everything.
        assert_eq!(pm.get_url_for_endpoint("http://nope.invalid:9000"), None);
        assert_eq!(pm.get_url_for_endpoint("http://10.0.0.9:9000"), None);
    }

    #[test]
    fn port_of_defaults_by_scheme() {
        assert_eq!(super::port_of("http://h:9000"), 9000);
        assert_eq!(super::port_of("http://h"), 80);
        assert_eq!(super::port_of("https://h"), 443);
        assert_eq!(super::port_of("http://h:9000/peer/earn"), 9000);
    }

    #[test]
    fn test_determine_authority_single_broker() {
        let pm = PeerManager::new(Some("10.0.0.1"), &[], 9000, "key".into(), true);
        // Single broker → always Standalone (total_brokers <= 1)
        match pm.determine_authority("9000000001") {
            Authority::Standalone => {}
            other => panic!("Expected Standalone, got {:?}", other),
        }
    }

    #[test]
    fn test_determine_authority_disabled() {
        let pm = PeerManager::new(
            Some("10.0.0.1"),
            &["10.0.0.2:3960".to_string()],
            9000,
            "key".into(),
            false,
        );
        match pm.determine_authority("9000000001") {
            Authority::Standalone => {}
            other => panic!("Expected Standalone, got {:?}", other),
        }
    }

    // Spin up a throwaway axum "peer" on an ephemeral port and return its base URL.
    async fn spawn_mock_peer(app: axum::Router) -> String {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            let _ = axum::serve(listener, app).await;
        });
        format!("http://{}", addr)
    }

    #[tokio::test]
    async fn test_async_peer_client_happy_path_roundtrip() {
        use axum::{
            routing::{get, post},
            Json, Router,
        };
        use serde_json::json;

        let app = Router::new()
            .route(
                "/peer/balance",
                get(|| async { Json(json!({ "user_id": "u1", "balance": 7.25 })) }),
            )
            .route(
                "/peer/reserve",
                post(|| async {
                    Json(json!({ "reservation_id": "res-xyz", "balance_before": 12.5 }))
                }),
            )
            .route(
                "/peer/commit",
                post(|| async { Json(json!({ "balance_after": 3.0 })) }),
            )
            .route(
                "/peer/cancel",
                post(|| async { Json(json!({ "ok": true })) }),
            );
        let base = spawn_mock_peer(app).await;

        let http = reqwest::Client::new();
        let client = PeerClient::new(base, "k".to_string());

        assert_eq!(client.get_balance_async(&http, "u1").await.unwrap(), 7.25);

        let r = client
            .reserve_async(&http, "u1", 2.0, "req-1")
            .await
            .unwrap();
        assert_eq!(r.reservation_id, "res-xyz");
        assert_eq!(r.balance_before, 12.5);

        let c = client.commit_async(&http, "res-xyz", 1.5).await.unwrap();
        assert_eq!(c.balance_after, 3.0);

        client.cancel_async(&http, "res-xyz").await.unwrap();

        assert!(
            client.is_reachable(),
            "successful calls should mark the peer reachable"
        );
    }

    #[tokio::test]
    async fn test_async_peer_reserve_error_surfaces_peer_message_and_code() {
        use axum::http::StatusCode;
        use axum::{routing::post, Json, Router};
        use serde_json::json;

        let app = Router::new().route(
            "/peer/reserve",
            post(|| async {
                (
                    StatusCode::PAYMENT_REQUIRED,
                    Json(
                        json!({ "error": "Insufficient credits", "code": "INSUFFICIENT_CREDITS" }),
                    ),
                )
            }),
        );
        let base = spawn_mock_peer(app).await;

        let http = reqwest::Client::new();
        let client = PeerClient::new(base, "k".to_string());

        let err = client
            .reserve_async(&http, "u1", 999.0, "req-2")
            .await
            .unwrap_err();
        assert!(
            err.contains("Insufficient credits"),
            "should surface peer error, got: {err}"
        );
        assert!(
            err.contains("402"),
            "should include the status code, got: {err}"
        );
    }

    #[tokio::test]
    async fn test_async_peer_unreachable_is_error_not_panic() {
        let http = reqwest::Client::new();
        // Port 1 is privileged/closed — connect fails fast.
        let client = PeerClient::new("http://127.0.0.1:1".to_string(), "k".to_string());
        let err = client.get_balance_async(&http, "u1").await.unwrap_err();
        assert!(err.contains("unreachable"), "got: {err}");
        assert!(
            !client.is_reachable(),
            "failed call should mark peer unreachable"
        );
    }
}

/// Finding 1: `load_persisted_from` used to discard the fingerprint it had just
/// read off disk, leaving the identity cache empty for up to ~60s after every
/// restart — during which the earn path resolved no peer and every remote job
/// was executed for free.
#[cfg(test)]
mod persisted_fingerprint_seeding_tests {
    use super::PeerManager;

    fn manager() -> PeerManager {
        PeerManager::new(
            Some("10.13.13.2"),
            &[] as &[String],
            9000,
            "k".to_string(),
            true,
        )
    }

    #[test]
    fn persisted_fingerprint_is_seeded_on_load_without_any_health_probe() {
        let dir = std::env::temp_dir().join(format!("zc-peer-persist-{}", uuid::Uuid::new_v4()));
        std::fs::create_dir_all(&dir).unwrap();

        // Round-trip through the real writer so the on-disk shape is production's.
        // Seed the BARE form here on purpose: canonicalization must not depend on
        // what the caller happened to pass (finding 3).
        let writer = manager();
        writer.register_peer("http://10.13.13.7:9000".to_string());
        writer.set_peer_fingerprint("http://10.13.13.7:9000", "aaaaaaaaaaaaaaaa");
        assert_eq!(
            writer.peer_fingerprint("http://10.13.13.7:9000").as_deref(),
            Some("zc://node-aaaaaaaaaaaaaaaa"),
            "a bare fp must be canonicalized at the write seam"
        );
        writer.persist_to(&dir);

        // Fresh manager = a restarted broker. No health probe runs.
        let restarted = manager();
        restarted.load_persisted_from(&dir);

        // Stored shape survives the disk round-trip as the canonical form, so
        // `/peers` and `/brokers` emit an addressable id straight after restart.
        assert_eq!(
            restarted
                .peer_fingerprint("http://10.13.13.7:9000")
                .as_deref(),
            Some("zc://node-aaaaaaaaaaaaaaaa"),
            "reloaded identity must be canonical `zc://node-<fp>`, not bare hex"
        );
        // Money path unchanged: `get_url_for_fingerprint` strips BOTH sides, so
        // the canonical stored value resolves exactly as a bare one did.
        assert_eq!(
            restarted
                .get_url_for_fingerprint("aaaaaaaaaaaaaaaa")
                .as_deref(),
            Some("http://10.13.13.7:9000"),
            "earn path must resolve a persisted peer immediately after restart"
        );

        std::fs::remove_dir_all(&dir).ok();
    }

    /// Finding 3: every writer of the identity cache must converge on ONE shape.
    /// The three writers are `check_health` (passes the `zc://node-<fp>`
    /// `broker_id` verbatim), `load_persisted_from` (passes whatever was on
    /// disk) and `discovery::admit_verified_peer_with` (passes a bare fp). Since
    /// all three now go through `PeerClient::set_fingerprint`, feeding either
    /// shape must yield byte-identical stored state.
    #[test]
    fn all_writers_converge_on_the_canonical_shape() {
        let bare = "1234567890abcdef";
        let canonical = format!("zc://node-{bare}");

        // Shape the eager seeding paths supply.
        let seeded = manager();
        seeded.register_peer("http://10.13.13.11:9000".to_string());
        seeded.set_peer_fingerprint("http://10.13.13.11:9000", bare);

        // Shape `check_health` supplies (`health.broker_id`), which it now hands
        // to the same `set_fingerprint` seam.
        let probed = manager();
        probed.register_peer("http://10.13.13.11:9000".to_string());
        probed.set_peer_fingerprint("http://10.13.13.11:9000", &canonical);

        let a = seeded.peer_fingerprint("http://10.13.13.11:9000");
        let b = probed.peer_fingerprint("http://10.13.13.11:9000");
        assert_eq!(
            a, b,
            "seeding and health-probe writers disagree on the stored shape"
        );
        assert_eq!(a.as_deref(), Some(canonical.as_str()));

        // A blank identity must stay absent, never become the matches-nothing
        // string `zc://node-`.
        let blank = manager();
        blank.register_peer("http://10.13.13.12:9000".to_string());
        blank.set_peer_fingerprint("http://10.13.13.12:9000", "");
        assert_eq!(blank.peer_fingerprint("http://10.13.13.12:9000"), None);
        blank.set_peer_fingerprint("http://10.13.13.12:9000", "zc://node-");
        assert_eq!(blank.peer_fingerprint("http://10.13.13.12:9000"), None);
    }

    #[test]
    fn persisted_peer_without_a_fingerprint_is_registered_but_unresolvable() {
        // A peer persisted before it was ever probed has fp == "" — it must be
        // registered as a hint, but must never resolve (an empty fingerprint
        // matching anything would mis-route money).
        let dir = std::env::temp_dir().join(format!("zc-peer-persist-{}", uuid::Uuid::new_v4()));
        std::fs::create_dir_all(&dir).unwrap();
        let writer = manager();
        writer.register_peer("http://10.13.13.9:9000".to_string());
        writer.persist_to(&dir);

        let restarted = manager();
        restarted.load_persisted_from(&dir);
        assert!(restarted
            .peer_urls()
            .contains(&"http://10.13.13.9:9000".to_string()));
        assert_eq!(restarted.peer_fingerprint("http://10.13.13.9:9000"), None);
        assert_eq!(restarted.get_url_for_fingerprint(""), None);

        std::fs::remove_dir_all(&dir).ok();
    }
}

/// Fix C + Fix F: roster-derived peers are deduped by IDENTITY and are
/// evictable, while `ZAKURO_PEERS`-configured peers are neither.
///
/// These are the two properties with no coverage before this change, and the
/// two most likely to regress: the first falsifies `zc brokers`' acceptance
/// criterion when it breaks, and the second leaks a PeerClient per address
/// ever advertised, silently, until the process restarts.
#[cfg(test)]
mod roster_peer_lifecycle_tests {
    use super::PeerManager;
    use std::collections::HashSet;

    const CFG_URL: &str = "http://zc-broker-1.zc-broker.zakuro-staging.svc.cluster.local:9000";
    const ROSTER_URL: &str = "http://10.13.13.7:9000";
    const FP: &str = "aabbccddeeff0011";

    /// A PeerManager with one ZAKURO_PEERS-configured peer whose identity has
    /// already been resolved by a health probe.
    fn with_configured_peer() -> PeerManager {
        let pm = PeerManager::new(
            Some("10.13.13.5"),
            &[CFG_URL.to_string()],
            9000,
            "k".into(),
            true,
        );
        pm.set_peer_fingerprint(CFG_URL, FP);
        pm
    }

    // --- Fix C: one broker, one PeerClient -----------------------------------

    /// The same physical broker under two spellings must resolve to ONE
    /// registered peer. Before the fix, `merge_peer_addresses` compared URL
    /// strings, saw two distinct entries and registered both — two health
    /// probes and two identical `zc://node-<fp>` rows for one broker.
    #[test]
    fn configured_peer_is_found_by_fingerprint_under_a_different_spelling() {
        let pm = with_configured_peer();
        assert_eq!(
            pm.configured_url_for_fingerprint(FP).as_deref(),
            Some(CFG_URL),
            "roster spelling must resolve to the already-configured broker"
        );
        assert_eq!(pm.peer_count(), 1);
    }

    /// The canonical `zc://node-<fp>` form and the bare fingerprint must
    /// resolve identically, so the dedup cannot be defeated by shape.
    #[test]
    fn fingerprint_lookup_accepts_both_bare_and_canonical_forms() {
        let pm = with_configured_peer();
        assert_eq!(
            pm.configured_url_for_fingerprint(&format!("zc://node-{FP}"))
                .as_deref(),
            Some(CFG_URL)
        );
        assert_eq!(pm.configured_url_for_fingerprint("").as_deref(), None);
    }

    /// An unresolved fingerprint must read as "unknown", never as a match —
    /// a reachability failure must not silently alias two brokers together.
    #[test]
    fn unprobed_configured_peer_has_no_fingerprint_match() {
        let pm = PeerManager::new(
            Some("10.13.13.5"),
            &[CFG_URL.to_string()],
            9000,
            "k".into(),
            true,
        );
        assert_eq!(pm.configured_url_for_fingerprint(FP), None);
    }

    /// A roster-derived peer must NOT satisfy the configured-peer lookup;
    /// otherwise the dedup would resolve against a duplicate it created
    /// itself and could never converge on removing it.
    #[test]
    fn roster_derived_peer_does_not_answer_the_configured_lookup() {
        let pm = PeerManager::new(Some("10.13.13.5"), &[], 9000, "k".into(), true);
        pm.register_roster_peer(ROSTER_URL.to_string());
        pm.set_peer_fingerprint(ROSTER_URL, FP);

        assert!(pm.is_roster_derived(ROSTER_URL));
        assert_eq!(pm.configured_url_for_fingerprint(FP), None);
    }

    // --- Fix F: roster-derived peers are evictable ---------------------------

    /// The core of Fix F: a peer that has left the roster is dropped, not
    /// health-checked forever.
    #[test]
    fn roster_peer_absent_from_the_roster_is_evicted() {
        let pm = PeerManager::new(Some("10.13.13.5"), &[], 9000, "k".into(), true);
        pm.register_roster_peer(ROSTER_URL.to_string());
        assert_eq!(pm.peer_count(), 1);

        let evicted = pm.retain_roster_peers(&HashSet::new());

        assert_eq!(evicted, vec![ROSTER_URL.to_string()]);
        assert_eq!(pm.peer_count(), 0, "roster peer was never removed");
        assert!(pm.peer_urls().is_empty());
    }

    /// Still on the roster => still registered. Eviction must not churn the
    /// peer set every tick.
    #[test]
    fn roster_peer_still_on_the_roster_is_kept() {
        let pm = PeerManager::new(Some("10.13.13.5"), &[], 9000, "k".into(), true);
        pm.register_roster_peer(ROSTER_URL.to_string());

        let keep: HashSet<String> = [ROSTER_URL.to_string()].into_iter().collect();
        assert!(pm.retain_roster_peers(&keep).is_empty());
        assert_eq!(pm.peer_count(), 1);
    }

    /// THE safety property: a `ZAKURO_PEERS` peer is the cluster-native
    /// fallback the fleet runs on today. An empty roster — the normal state
    /// before any node advertises an endpoint — must leave it completely
    /// untouched.
    #[test]
    fn configured_peer_is_never_evicted_by_an_empty_roster() {
        let pm = with_configured_peer();

        let evicted = pm.retain_roster_peers(&HashSet::new());

        assert!(
            evicted.is_empty(),
            "evicted a ZAKURO_PEERS peer: {evicted:?}"
        );
        assert_eq!(pm.peer_count(), 1);
        assert_eq!(pm.peer_urls(), vec![CFG_URL.to_string()]);
    }

    /// The guard lives in the removal primitive itself, so even a direct
    /// call naming a configured peer cannot drop it.
    #[test]
    fn remove_roster_peer_refuses_a_configured_peer() {
        let pm = with_configured_peer();

        assert!(!pm.remove_roster_peer(CFG_URL));
        assert!(!pm.remove_roster_peer("http://never-registered:9000"));
        assert_eq!(pm.peer_count(), 1);
    }

    /// Provenance only ever hardens toward "permanent": the roster naming a
    /// URL that is already configured must not make it evictable.
    #[test]
    fn roster_cannot_downgrade_a_configured_peer_to_evictable() {
        let pm = with_configured_peer();

        pm.register_roster_peer(CFG_URL.to_string());

        assert!(!pm.is_roster_derived(CFG_URL));
        assert!(pm.retain_roster_peers(&HashSet::new()).is_empty());
        assert_eq!(pm.peer_count(), 1);
    }

    /// Provenance must survive a restart. Without persisting the flag, a
    /// roster-derived peer reloads as configured and becomes permanently
    /// un-evictable — re-opening the very leak Fix F closes.
    #[test]
    fn roster_provenance_survives_persist_and_reload() {
        let dir = std::env::temp_dir().join(format!(
            "zc2-roster-provenance-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));

        let pm = PeerManager::new(
            Some("10.13.13.5"),
            &[CFG_URL.to_string()],
            9000,
            "k".into(),
            true,
        );
        pm.register_roster_peer(ROSTER_URL.to_string());
        pm.persist_to(&dir);

        let pm2 = PeerManager::new(
            Some("10.13.13.5"),
            &[CFG_URL.to_string()],
            9000,
            "k".into(),
            true,
        );
        pm2.load_persisted_from(&dir);

        assert!(pm2.is_roster_derived(ROSTER_URL), "lost roster provenance");
        assert!(
            !pm2.is_roster_derived(CFG_URL),
            "configured peer became evictable"
        );

        let evicted = pm2.retain_roster_peers(&HashSet::new());
        assert_eq!(evicted, vec![ROSTER_URL.to_string()]);
        assert_eq!(pm2.peer_urls(), vec![CFG_URL.to_string()]);

        let _ = std::fs::remove_dir_all(&dir);
    }

    /// A `peers.json` from an older zc has no `roster_derived` key. Those
    /// entries must reload as CONFIGURED (never evicted) — an upgrade must
    /// not start silently dropping peers.
    #[test]
    fn legacy_peers_json_without_the_flag_reloads_as_configured() {
        let dir = std::env::temp_dir().join(format!(
            "zc2-legacy-peers-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        std::fs::create_dir_all(&dir).unwrap();
        std::fs::write(
            dir.join("peers.json"),
            format!(r#"[{{"fp":"{FP}","url":"{ROSTER_URL}","last_seen":1}}]"#),
        )
        .unwrap();

        let pm = PeerManager::new(Some("10.13.13.5"), &[], 9000, "k".into(), true);
        pm.load_persisted_from(&dir);

        assert_eq!(pm.peer_count(), 1);
        assert!(!pm.is_roster_derived(ROSTER_URL));
        assert!(pm.retain_roster_peers(&HashSet::new()).is_empty());

        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Eviction must clear every index keyed on the peer's URL, not just the
    /// client map, or the next registration inherits a stale QUIC port.
    #[test]
    fn eviction_clears_the_quic_port_index() {
        let pm = PeerManager::new(Some("10.13.13.5"), &[], 9000, "k".into(), true);
        pm.register_roster_peer(ROSTER_URL.to_string());
        pm.set_quic_port(ROSTER_URL, 9001);
        assert_eq!(pm.get_quic_port(ROSTER_URL), 9001);

        pm.retain_roster_peers(&HashSet::new());

        assert_eq!(pm.get_quic_port(ROSTER_URL), 0, "stale QUIC port survived");
    }
}