zc2 0.0.27

//! Worker discovery for the P2P broker.
//!
//! Discovery modes:
//! 1. WireGuard mode: Scans the WireGuard subnet (10.13.13.0/24 or 100.x.x.x)
//! 2. Local mode: Falls back to localhost when WireGuard unavailable
//! 3. Manual: Workers can always register via /workers endpoint

use std::net::TcpStream;
use std::sync::Arc;
use std::thread;
use std::time::Duration;

use super::worker::{HardwareInfo, WorkerPricing, WorkerRegistration, WorkerResources};
use super::BrokerState;
use colored::Colorize;
use serde::{Deserialize, Serialize};

/// Sanitize a peer-advertised price: accept only a finite value in [0, max];
/// otherwise fall back to `default`. A malicious/buggy peer must not be able to
/// inject negative/NaN/absurd pricing into routing or billing (audit M4).
fn sanitize_price(raw: Option<f64>, default: f64, max: f64) -> f64 {
    match raw {
        Some(v) if v.is_finite() && v >= 0.0 && v <= max => v,
        _ => default,
    }
}

/// Discovery mode
#[derive(Debug, Clone, PartialEq)]
pub enum DiscoveryMode {
    /// WireGuard network available - scan subnet
    WireGuard { subnet: String },
    /// No WireGuard - scan localhost only
    Local,
}

/// Discovery configuration
#[derive(Debug, Clone)]
pub struct DiscoveryConfig {
    /// WireGuard subnet to scan (default: 10.13.13.0/24)
    pub subnet: String,
    /// Primary port workers listen on (default: 3960)
    pub worker_port: u16,
    /// Additional explicit ports to scan for workers
    pub extra_ports: Vec<u16>,
    /// Scan all localhost ports in this inclusive range [start, end].
    /// When set, discovery replaces the fixed extra_ports list with a full
    /// range scan so workers started on arbitrary ports are discovered.
    pub scan_port_range: Option<(u16, u16)>,
    /// Discovery interval in seconds
    pub interval_secs: u64,
    /// Enable active network scanning
    pub enable_scan: bool,
    /// Enable DNS-based discovery
    pub enable_dns: bool,
    /// Explicit peer addresses to probe (from ZAKURO_PEERS env var)
    /// Format: comma-separated "ip:port" or just "ip" (uses worker_port)
    pub peers: Vec<String>,
    /// Addresses of workers this broker OWNS that are not reachable on
    /// loopback (from ZAKURO_WORKERS). See `parse_local_workers`.
    pub local_workers: Vec<(String, u16)>,
}

impl Default for DiscoveryConfig {
    fn default() -> Self {
        let peers = std::env::var("ZAKURO_PEERS")
            .unwrap_or_default()
            .split(',')
            .map(|s| s.trim().to_string())
            .filter(|s| !s.is_empty())
            .collect();

        // Respect ZAKURO_WORKER_PORT env var for custom port configuration
        let worker_port = std::env::var("ZAKURO_WORKER_PORT")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(3960);

        // Allow ZAKURO_SCAN_RANGE="3960-3999" for environment-driven range scan
        let scan_port_range = std::env::var("ZAKURO_SCAN_RANGE").ok().and_then(|v| {
            let parts: Vec<&str> = v.splitn(2, '-').collect();
            if parts.len() == 2 {
                let start = parts[0].parse::<u16>().ok()?;
                let end = parts[1].parse::<u16>().ok()?;
                Some((start, end))
            } else {
                None
            }
        });

        let local_workers = parse_local_workers(
            &std::env::var("ZAKURO_WORKERS").unwrap_or_default(),
            worker_port,
        );

        Self {
            subnet: "10.13.13".to_string(),
            worker_port,
            extra_ports: vec![3961, 3962], // Common alternative ports
            scan_port_range,
            interval_secs: std::env::var("ZAKURO_SCAN_INTERVAL")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(15), // Must be less than worker_timeout (30s)
            enable_scan: true,
            enable_dns: true,
            peers,
            local_workers,
        }
    }
}

/// Worker discovery service
pub struct Discovery {
    config: DiscoveryConfig,
    state: Arc<BrokerState>,
    mode: DiscoveryMode,
}

impl Discovery {
    /// Create a new discovery service with automatic mode detection
    pub fn new(config: DiscoveryConfig, state: Arc<BrokerState>) -> Self {
        let mode = detect_discovery_mode(&config.subnet);
        Self {
            config,
            state,
            mode,
        }
    }

    /// Get the current discovery mode
    pub fn mode(&self) -> &DiscoveryMode {
        &self.mode
    }

    /// Run discovery loop (blocking)
    pub fn run(&self, verbose: bool) {
        // Log initial mode
        if verbose {
            match &self.mode {
                DiscoveryMode::WireGuard { subnet } => {
                    println!(
                        "  {} WireGuard mode (subnet: {}.0/24)",
                        "[DISCOVERY]".cyan(),
                        subnet
                    );
                }
                DiscoveryMode::Local => {
                    println!(
                        "  {} Local mode (scanning localhost)",
                        "[DISCOVERY]".yellow()
                    );
                }
            }
            if !self.config.peers.is_empty() {
                println!(
                    "  {} Peers: {}",
                    "[DISCOVERY]".cyan(),
                    self.config.peers.join(", ")
                );
            }
        }

        // Initial scan immediately
        if self.config.enable_scan {
            self.discover_workers(verbose);
        }

        // Then periodic scans
        loop {
            thread::sleep(Duration::from_secs(self.config.interval_secs));
            if self.config.enable_scan {
                self.discover_workers(verbose);
            }
        }
    }

    /// Discover workers based on current mode
    ///
    /// Design model: a broker OWNS and manages ONLY its own local workers.
    /// The mesh is a set of BROKERS talking P2P; cross-node workers are not a
    /// broker's concern — only its peers (other brokers) are. Worker
    /// auto-discovery is therefore constrained to localhost. The former
    /// 253-host `10.13.13.0/24:3960` subnet sweep (`scan_subnet`) registered
    /// remote mesh workers as LOCAL (source_node=None), which made every
    /// broker show other brokers' workers as its own — that call is
    /// deliberately removed here. Peer brokers are still discovered/probed
    /// via `scan_peers` (which registers PEER BROKERS in `PeerManager`, not
    /// local workers), and a broker that legitimately needs a peer's worker
    /// list can still fetch it via the authenticated `/peer/workers` sync,
    /// which always stamps `source_node` (never local).
    fn discover_workers(&self, verbose: bool) {
        // Always scan localhost first (finds local worker in any mode)
        self.scan_localhost(verbose);

        // Then any worker addresses the operator named explicitly. Needed
        // whenever this broker's own worker is not on loopback -- see the
        // sidecar-namespace case in `try_register_worker`.
        for (host, port) in self.config.local_workers.clone() {
            self.try_register_worker(&host, port, verbose);
        }

        // Probe explicit peers early — registers peer BROKERS (PeerManager)
        // and syncs their worker lists via the peer-scoped /peer/workers API
        // (source_node always set); never registers a remote worker as local.
        self.scan_peers(verbose);

        // NOTE: subnet worker scanning (`scan_subnet`) is intentionally NOT
        // invoked here — see doc comment above. The function is retained
        // (dead_code-allowed) only so existing unit tests / callers that
        // exercise it directly (if any) keep compiling; it is not part of
        // the local-worker discovery path.
    }

    /// Probe explicit peer addresses (workers + peer brokers), then sync
    /// worker lists from any OTHER broker `PeerManager` already knows about.
    ///
    /// Two independent peer lists exist on this broker: `self.config.peers`
    /// (`DiscoveryConfig`, sourced solely from `ZAKURO_PEERS`) and the live
    /// set inside `self.state.peer_manager`, which also gains entries from
    /// mesh-subnet/localhost discovery at startup (`mod.rs`
    /// `BrokerState::with_config`) and from gossip admission at runtime
    /// (`run_discovery_round` / `admit_verified_peer_with`). Those
    /// dynamically-discovered brokers used to be completely invisible to
    /// worker discovery: this loop only ever walked `self.config.peers`, so
    /// `/peer/workers` was never fetched from them even though the broker
    /// itself was reachable and logged as "[P2P] Registered peer broker".
    /// Fixed by also reading `peer_manager.peer_urls()` — on every tick, not
    /// just once at startup, so a peer admitted later via gossip is picked
    /// up without a restart.
    fn scan_peers(&self, verbose: bool) {
        let broker_port = self.state.config.port;
        let mut explicit_hosts = std::collections::HashSet::new();

        for peer in &self.config.peers {
            let (host, port) = if let Some((h, p)) = peer.rsplit_once(':') {
                (h.to_string(), p.parse().unwrap_or(self.config.worker_port))
            } else {
                (peer.clone(), self.config.worker_port)
            };
            if host.is_empty() {
                continue;
            }
            // Every explicit peer is scanned for /peer/workers at
            // (host, broker_port) below regardless of which branch runs —
            // record that so the dynamic-peer merge below doesn't re-scan it.
            explicit_hosts.insert((host.clone(), broker_port));

            if port == broker_port {
                // Peer specified as broker address — fetch workers via broker-to-broker API
                self.try_register_peer_broker(&host, broker_port, verbose);
                self.fetch_workers_from_peer_broker(&host, broker_port, verbose);
            } else {
                // Peer specified as worker port — direct probe (same-network workers)
                self.try_register_worker(&host, port, verbose);
                // Also register the peer broker for credit operations
                self.try_register_peer_broker(&host, broker_port, verbose);
                self.fetch_workers_from_peer_broker(&host, broker_port, verbose);
            }
        }

        // Read the PeerManager's live peer list (not a startup snapshot) —
        // it already excludes this broker's own address (see
        // `PeerManager::new_with_node_key`), so no self-filtering is needed
        // here beyond the explicit-peer dedup.
        let dynamic_urls = self.state.peer_manager.peer_urls();
        for (host, port) in resolve_dynamic_peer_broker_addrs(&dynamic_urls, &explicit_hosts) {
            self.try_register_peer_broker(&host, port, verbose);
            self.fetch_workers_from_peer_broker(&host, port, verbose);
        }
    }

    /// Fetch workers from a peer broker via /peer/workers and register them locally.
    /// Worker URIs are rewritten from 127.0.0.1:PORT to PEER_IP:PORT so they are
    /// treated as remote (non-local) workers and billed accordingly.
    fn fetch_workers_from_peer_broker(&self, host: &str, broker_port: u16, verbose: bool) {
        let peer_key = self.state.peer_manager.peer_key();
        let url = format!("http://{}:{}/peer/workers", host, broker_port);

        // Route through the vpn sidecar's CONNECT proxy for mesh peers (10.13.13.0/24)
        // — a broker reaching the mesh via the proxy (not on the mesh netns itself)
        // otherwise has no route to peer brokers.
        let mut cfg = ureq::Agent::config_builder().timeout_global(Some(Duration::from_secs(3)));
        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());
        let result = agent.get(&url).header("X-Peer-Key", peer_key).call();

        let body = match result {
            Ok(resp) => match resp.into_body().read_to_string() {
                Ok(s) => s,
                Err(_) => return,
            },
            Err(_) => return,
        };

        let json: serde_json::Value = match serde_json::from_str(&body) {
            Ok(v) => v,
            Err(_) => return,
        };

        let workers = match json["workers"].as_array() {
            Some(arr) => arr,
            None => return,
        };

        for w in workers {
            let name = w["name"].as_str().unwrap_or("").to_string();
            let uri = w["uri"].as_str().unwrap_or("").to_string();
            let worker_type = w["worker_type"].as_str().unwrap_or("zakuro").to_string();

            if name.is_empty() || uri.is_empty() {
                continue;
            }

            // Owning node handle, e.g. `zc://node-i9` — the server side
            // (worker_to_info in server.rs) always populates this, even for a
            // peer's own local workers, so it is never empty for a well-formed
            // peer response. Fall back to a synthetic-but-non-empty handle
            // rather than silently degrading to a "local" (None) worker if a
            // malformed/older peer omits it — a peer worker must never be
            // mistaken for a local one downstream.
            let node_field = w["node"].as_str().unwrap_or("");
            let source_node = {
                let bare = strip_zc_node(node_field);
                if bare.is_empty() {
                    format!("node-unknown-{}", host)
                } else {
                    bare.to_string()
                }
            };

            // Rewrite uri: http://127.0.0.1:PORT  →  http://PEER_IP:PORT
            let rewritten_uri = if let Some(rest) = uri.strip_prefix("http://127.0.0.1:") {
                format!(
                    "http://{}:{}",
                    host,
                    rest.split('/').next().unwrap_or("3960")
                )
            } else if let Some(rest) = uri.strip_prefix("http://localhost:") {
                format!(
                    "http://{}:{}",
                    host,
                    rest.split('/').next().unwrap_or("3960")
                )
            } else {
                uri.clone() // already has correct host
            };

            // If already registered, refresh heartbeat to keep it alive, then skip.
            // Dedup by KEY-DERIVED identity — (source node fingerprint, name/slot) —
            // not bare name. On a fleet where every host's hostname is "lxd", every
            // worker is named "worker-lxd"; matching on name alone would conflate
            // node A's worker-lxd with node B's worker-lxd (or with this broker's
            // own local worker-lxd, which must never match a peer at all).
            let existing = self.state.workers.list();
            if let Some(known) = existing
                .iter()
                .find(|e| is_same_peer_worker(e, &name, &rewritten_uri, &source_node))
            {
                self.state.workers.refresh_heartbeat(&known.id);
                continue;
            }

            let (provider_type, served_models, price_per_mtok) = provider_fields_from_json(w);

            let registration = WorkerRegistration {
                name: name.clone(),
                uri: rewritten_uri.clone(),
                worker_type,
                resources: WorkerResources {
                    cpus_available: w["cpus_available"].as_f64().unwrap_or(1.0),
                    cpus_total: w["cpus_total"].as_f64().unwrap_or(1.0),
                    memory_available: (w["memory_available_gib"].as_f64().unwrap_or(1.0)
                        * 1024.0
                        * 1024.0
                        * 1024.0) as u64,
                    memory_total: (w["memory_total_gib"].as_f64().unwrap_or(1.0)
                        * 1024.0
                        * 1024.0
                        * 1024.0) as u64,
                    gpus_available: w["gpus_available"].as_u64().unwrap_or(0) as u32,
                    gpus_total: w["gpus_total"].as_u64().unwrap_or(0) as u32,
                },
                pricing: WorkerPricing {
                    price_per_hour: sanitize_price(w["price_per_hour"].as_f64(), 3.6, 1000.0),
                    min_charge: sanitize_price(w["min_charge"].as_f64(), 0.001, 100.0),
                },
                tags: vec![],
                max_timeout_secs: 0.0,
                hardware: HardwareInfo {
                    cpu_model: w["cpu_model"].as_str().map(|s| s.to_string()),
                    gpu_model: w["gpu_model"].as_str().map(|s| s.to_string()),
                    gpu_vram_gb: w["gpu_vram_gb"].as_u64().map(|v| v as u32),
                    storage_gb: w["storage_gb"].as_u64().map(|v| v as u32),
                },
                wireguard_ip: Some(host.to_string()),
                is_docker: w["is_docker"].as_bool(),
                // Owning node, carried on /peer/workers as `zc://node-…`. Always
                // Some(..) for a synced peer worker — see `source_node` computed
                // above (falls back to a synthetic-but-non-empty handle rather
                // than None, so a peer worker can never be mistaken for local).
                source_node: Some(source_node.clone()),
                // Peer-synced worker: never this broker's own.
                explicit_local: false,
                provider_type,
                served_models,
                price_per_mtok,
            };

            let worker = self.state.workers.register(registration);
            // The source broker's `node` handle is `zc://node-{fingerprint}` (or bare
            // `node-{fingerprint}`, per strip_zc_node above); stamp that fingerprint onto
            // the locally-registered peer worker so its own zc_uri() (`zc://worker-{node_fp}-
            // {slot}`) is well-formed. Without this, node_fp stays "" (Worker::new default)
            // and the uri comes out `zc://worker--{slot}` (double dash).
            if let Some(fp) = peer_node_fp(&worker) {
                self.state.workers.set_node_fp(&worker.id, &fp);
            }
            if verbose {
                println!(
                    "  {} Discovered peer worker {} at {} (via broker {}:{})",
                    "[DISCOVERY]".cyan(),
                    worker.name,
                    rewritten_uri,
                    host,
                    broker_port,
                );
            }
        }
    }

    /// Try to register a peer broker at the given address.
    /// Probes /peer/health and registers in PeerManager if alive.
    fn try_register_peer_broker(&self, host: &str, port: u16, verbose: bool) {
        let addr = format!("{}:{}", host, port);

        // Resolve hostname to socket address (supports both IPs and hostnames)
        use std::net::ToSocketAddrs;
        let sock_addr: std::net::SocketAddr = match addr.to_socket_addrs() {
            Ok(mut addrs) => match addrs.next() {
                Some(a) => a,
                None => return,
            },
            Err(_) => return,
        };

        if TcpStream::connect_timeout(&sock_addr, Duration::from_millis(200)).is_err() {
            return;
        }

        let base_url = format!("http://{}:{}", host, port);
        self.state.peer_manager.register_peer(base_url.clone());

        // Check health
        if let Some(client) = self.state.peer_manager.get_client(&base_url) {
            if client.check_health() && verbose {
                println!(
                    "  {} Peer broker alive at {}:{}",
                    "[DISCOVERY]".cyan(),
                    host,
                    port
                );
            }
        }
    }

    /// Scan localhost for workers — parallel TCP probing for fast discovery.
    ///
    /// All ports are probed concurrently: TCP SYN is sent to every port
    /// simultaneously, so the total time is bounded by a single connect
    /// timeout rather than N × timeout for sequential scanning.
    fn scan_localhost(&self, verbose: bool) {
        // Build the port list to scan
        let ports: Vec<u16> = if let Some((start, end)) = self.config.scan_port_range {
            (start..=end).collect()
        } else {
            let mut p = vec![self.config.worker_port];
            p.extend(&self.config.extra_ports);
            p
        };

        // Phase 1 — parallel TCP connect (50ms timeout, fast reject for closed ports)
        let open_ports: Vec<u16> = {
            let (tx, rx) = std::sync::mpsc::channel();
            let mut handles = Vec::with_capacity(ports.len());

            for port in &ports {
                let port = *port;
                let tx = tx.clone();
                handles.push(thread::spawn(move || {
                    let addr: std::net::SocketAddr = format!("127.0.0.1:{}", port).parse().unwrap();
                    if TcpStream::connect_timeout(&addr, Duration::from_millis(50)).is_ok() {
                        let _ = tx.send(port);
                    }
                }));
            }
            drop(tx); // close sender so rx.iter() terminates

            let mut open: Vec<u16> = rx.iter().collect();
            for h in handles {
                let _ = h.join();
            }
            open.sort_unstable();
            open
        };

        // Phase 2 — probe only the open ports (health + info) in parallel.
        // We use a thread per open port; on loopback each call is < 5ms.
        // Arc the state so each thread can register workers independently.
        let state = Arc::clone(&self.state);
        let config = self.config.clone();
        let mut probe_handles = Vec::with_capacity(open_ports.len());

        for port in open_ports {
            let state = Arc::clone(&state);
            let config = config.clone();
            probe_handles.push(thread::spawn(move || {
                let tmp = Discovery {
                    config,
                    state: Arc::clone(&state),
                    mode: DiscoveryMode::Local,
                };
                tmp.try_register_worker("127.0.0.1", port, verbose);
            }));
        }
        for h in probe_handles {
            let _ = h.join();
        }
    }

    /// Try to register a worker at the given address.
    ///
    /// Local-worker auto-discovery is constrained to localhost (see
    /// `is_loopback_host`): a broker owns and manages only its own local
    /// workers, never workers found elsewhere on the mesh. This gate is what
    /// makes the (still-present) explicit-peer worker-port probe in
    /// `scan_peers` a no-op for non-loopback hosts — cross-node workers are
    /// discovered, if at all, only via the authenticated `/peer/workers`
    /// sync, which stamps `source_node` and is never treated as local.
    fn try_register_worker(&self, host: &str, port: u16, verbose: bool) {
        // A broker owns only its own local workers, so auto-discovery stays
        // pinned to loopback. But a broker sharing a sidecar's network
        // namespace (`network_mode: container:<wireguard>`) reaches its OWN
        // worker on the host network only via the container gateway, e.g.
        // 172.17.0.1:3960 -- never loopback. That topology left /workers
        // empty and 503'd every job in staging on 2026-08-14. Naming an
        // address in ZAKURO_WORKERS is the operator asserting "this worker is
        // mine", so it bypasses the gate; nothing unnamed is ever adopted, so
        // the no-mesh-worker-as-local invariant still holds.
        if !is_loopback_host(host)
            && !is_explicit_local_worker(&self.config.local_workers, host, port)
        {
            return;
        }
        let addr = format!("{}:{}", host, port);

        // Resolve hostname to socket address (supports both IPs and hostnames)
        use std::net::ToSocketAddrs;
        let sock_addr = match addr.to_socket_addrs() {
            Ok(mut addrs) => match addrs.next() {
                Some(a) => a,
                None => return,
            },
            Err(_) => return,
        };

        // Connection check. Explicit peers (ZAKURO_PEERS) and subnet hosts may be
        // remote — reached over WireGuard/WireGuard, sometimes via a relay hub
        // across regions — so the round-trip can be well over 50ms. Use a
        // WAN-tolerant timeout; loopback/LAN still returns near-instantly.
        if TcpStream::connect_timeout(&sock_addr, Duration::from_millis(1500)).is_err() {
            return;
        }

        // Build canonical URI for dedup
        let uri = format!("http://{}:{}", host, port);

        // Check if already registered (by URI or localhost variants)
        let existing = self.state.workers.list();
        let existing_worker = existing.iter().find(|w| {
            w.uri == uri
                || (host == "127.0.0.1" && w.uri.contains("localhost"))
                || (host == "localhost" && w.uri.contains("127.0.0.1"))
        });

        if let Some(worker) = existing_worker {
            // Worker exists and TCP reachable — re-probe /info to get fresh resources
            if let Some(info) = self.probe_worker(host, port) {
                if let Some(resources) = info.resources {
                    self.state.workers.update_resources(
                        &worker.id,
                        resources,
                        info.hardware.unwrap_or_default(),
                    );
                } else {
                    self.state.workers.refresh_heartbeat(&worker.id);
                }
            } else {
                self.state.workers.refresh_heartbeat(&worker.id);
            }
            return;
        }

        // New endpoint — do full probe to get worker info
        if let Some(worker_info) = self.probe_worker(host, port) {
            // NOTE: this function directly TCP-probes a bare host:port (subnet scan,
            // localhost scan, or a ZAKURO_PEERS worker-port entry) — the /info
            // response it parses (`WorkerProbeResult`) carries no node-fingerprint,
            // only a plain `name`. On a fleet where every host's hostname is "lxd",
            // every worker reports name "worker-lxd", so merging on bare name here
            // would have the exact same cross-node collision hazard as the peer-sync
            // bug above: host A's `worker-lxd` and host B's `worker-lxd` are DIFFERENT
            // workers with different uris, and must never be merged into one registry
            // entry just because they share a name. The uri-based "already
            // registered" check above already covers the legitimate case (re-probing
            // the same host:port); there is no safe cross-host name-based dedup
            // available at this layer, so we no longer attempt one — every distinct
            // uri is registered as its own worker.
            {
                let registration = WorkerRegistration {
                    name: worker_info
                        .name
                        .unwrap_or_else(|| format!("worker-{}", host)),
                    uri,
                    worker_type: worker_info
                        .worker_type
                        .unwrap_or_else(|| "zakuro".to_string()),
                    resources: worker_info.resources.unwrap_or_default(),
                    pricing: worker_info.pricing.unwrap_or_default(),
                    tags: worker_info.tags.unwrap_or_default(),
                    max_timeout_secs: 0.0,
                    hardware: worker_info.hardware.unwrap_or_default(),
                    wireguard_ip: None,
                    is_docker: None,
                    source_node: None,
                    // Declared via ZAKURO_WORKERS => this broker's own worker
                    // even though it is not on loopback. Lets the router select
                    // it as local; see router::select_local_worker.
                    explicit_local: is_explicit_local_worker(
                        &self.config.local_workers,
                        host,
                        port,
                    ),
                    provider_type: Default::default(),
                    served_models: vec![],
                    price_per_mtok: 0.0,
                };

                let worker = self.state.workers.register(registration);
                // Local worker (source_node: None above): belongs to this broker's
                // node, so stamp its key-derived fingerprint.
                let worker = self
                    .state
                    .workers
                    .set_node_fp(&worker.id, &self.state.node_key.fingerprint())
                    .unwrap_or(worker);
                if verbose {
                    println!(
                        "  {} Discovered worker {} at {}",
                        "[DISCOVERY]".cyan(),
                        worker.name,
                        worker.uri
                    );
                }

                // Sync discovered worker immediately to dashboard
                if let Some(ref owner_id) = self.state.config.owner_user_id {
                    let node_name = self.state.config.node_name.as_deref();
                    let node_pubkey = self.state.node_key.public_b64();

                    // Prefer API sync if configured
                    if let (Some(ref api_url), Some(ref api_key)) =
                        (&self.state.config.api_url, &self.state.config.api_key)
                    {
                        match crate::broker::ledger::Ledger::sync_workers_via_api(
                            owner_id,
                            std::slice::from_ref(&worker),
                            api_url,
                            api_key,
                            node_name,
                            self.state.own_wireguard_ip.as_deref(),
                            Some(node_pubkey.as_str()),
                        ) {
                            Ok(outcome) => {
                                self.state.workers.apply_hub_prices(&outcome.prices);
                                if verbose {
                                    println!(
                                        "  [WORKER_SYNC] Worker {} synced to dashboard",
                                        worker.name
                                    );
                                }
                            }
                            Err(e) => {
                                eprintln!("  [WORKER_SYNC] Failed to sync {}: {}", worker.name, e);
                            }
                        }
                    }
                }
            }
        }
    }

    /// Probe a potential worker for its info
    /// Only returns Some if the worker has a valid /info endpoint with worker_type
    fn probe_worker(&self, ip: &str, port: u16) -> Option<WorkerProbeResult> {
        let health_url = format!("http://{}:{}/health", ip, port);

        // First check health
        match ureq::get(&health_url)
            .config()
            .timeout_global(Some(Duration::from_secs(2)))
            .build()
            .call()
        {
            Ok(response) if response.status().as_u16() == 200 => {
                // Must have a valid /info endpoint to be a zakuro worker
                let info_url = format!("http://{}:{}/info", ip, port);
                match ureq::get(&info_url)
                    .config()
                    .timeout_global(Some(Duration::from_secs(2)))
                    .build()
                    .call()
                {
                    Ok(info_response) if info_response.status().as_u16() == 200 => {
                        if let Ok(body) = info_response.into_body().read_to_string() {
                            if let Ok(info) = serde_json::from_str::<WorkerProbeResult>(&body) {
                                // Only accept workers with a recognized worker_type
                                if info.worker_type.is_some() {
                                    return Some(info);
                                }
                            }
                        }
                        None
                    }
                    _ => None, // No /info endpoint = not a zakuro worker
                }
            }
            _ => None,
        }
    }
}

/// One gossiped peer entry from a peer's `GET /peer/peers` response.
#[derive(Debug, Clone, serde::Deserialize)]
struct GossipEntry {
    fp: String,
    url: String,
    epoch: u64,
}

#[derive(Debug, Clone, serde::Deserialize)]
struct GossipResponse {
    peers: Vec<GossipEntry>,
}

/// Fetch a peer's full gossiped peer set via signed `GET /peer/peers`.
///
/// Dispatches over `agent` (build via `vpn::mesh_agent` for a mesh-routed
/// call). The response must carry `X-Node-Id`/`X-Node-Sig` (raw Ed25519 over
/// the JSON body, per `node_identity::verify_sig`); any missing/invalid
/// signature, or a signer whose fingerprint isn't `roster.fingerprint_authorized`,
/// drops the WHOLE response (empty result) rather than trusting a subset.
///
/// Addresses returned are HINTS ONLY — callers must re-probe before trusting
/// them (task 4). `epoch` (each entry's `last_seen`) is carried through
/// unused — a forward-compat hook for a future last-writer-wins merge; this
/// task always gossips the full peer set.
pub fn fetch_gossip(
    base_url: &str,
    roster: &super::roster_cache::RosterCache,
    agent: &ureq::Agent,
    node_key: &super::node_identity::NodeKey,
) -> Vec<(String, String, u64)> {
    let url = format!("{}/peer/peers", base_url);
    // `/peer/peers` is gated by `check_node_sig` (strict — see server.rs
    // handle_peer_peers): the request itself must carry a valid signed
    // X-Node-Id/Sig/Nonce/Ts, or the server 401s before this loop ever sees
    // a peer. Sign the same way `PeerClient::signed_get` does.
    let mut rb = agent.get(&url);
    for (k, v) in node_key.sign_headers("GET", "/peer/peers", b"") {
        rb = rb.header(k.as_str(), v.as_str());
    }
    let resp = match rb.call() {
        Ok(r) => r,
        Err(_) => return Vec::new(),
    };
    if resp.status().as_u16() != 200 {
        return Vec::new();
    }
    let node_id = resp
        .headers()
        .get("X-Node-Id")
        .and_then(|v| v.to_str().ok())
        .map(|s| s.to_string());
    let sig = resp
        .headers()
        .get("X-Node-Sig")
        .and_then(|v| v.to_str().ok())
        .map(|s| s.to_string());
    let (node_id, sig) = match (node_id, sig) {
        (Some(n), Some(s)) => (n, s),
        _ => return Vec::new(),
    };
    let body = match resp.into_body().read_to_string() {
        Ok(s) => s.into_bytes(),
        Err(_) => return Vec::new(),
    };
    verify_gossip_body(&body, &node_id, &sig, roster)
}

/// Pure verify+parse core of [`fetch_gossip`], split out for testability
/// without a live HTTP round-trip: verify `sig` over the raw `body` from
/// `node_id`, reject unless `node_id`'s fingerprint is roster-authorized,
/// then parse and return the gossiped entries.
fn verify_gossip_body(
    body: &[u8],
    node_id: &str,
    sig: &str,
    roster: &super::roster_cache::RosterCache,
) -> Vec<(String, String, u64)> {
    if !super::node_identity::verify_sig(node_id, body, sig) {
        return Vec::new();
    }
    let fp = match super::node_identity::fingerprint_of_pubkey_b64(node_id) {
        Some(fp) => fp,
        None => return Vec::new(),
    };
    if !roster.fingerprint_authorized(&fp) {
        return Vec::new();
    }
    let parsed: GossipResponse = match serde_json::from_slice(body) {
        Ok(p) => p,
        Err(_) => return Vec::new(),
    };
    parsed
        .peers
        .into_iter()
        .map(|e| (e.fp, e.url, e.epoch))
        .collect()
}

/// A broker's signed advertisement: its price + aggregated available
/// resources, served at `GET /peer/advert` (see `server::handle_peer_advert`).
///
/// `fp` is the advertiser's OWN node fingerprint — bound at verification time
/// (`verify_advert_body`) to the signer's fingerprint, so a broker can only
/// ever advertise as itself (no impersonation of another rostered broker).
/// `epoch` is the broker's monotonic seconds at build time (freshness stamp).
///
/// SECURITY: carries only fp + price + aggregated resources + epoch — never
/// per-worker IPs/addresses.
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct Advert {
    pub fp: String,
    pub price_per_hour: f64,
    pub resources: super::worker::BrokerResources,
    pub epoch: u64,
}

/// Fetch a peer's signed advertisement via `GET /peer/advert`.
///
/// Mirrors [`fetch_gossip`]: the outbound request is signed via
/// `node_key.sign_headers`, the response's `X-Node-Id`/`X-Node-Sig` are
/// verified over the raw body, the signer's fingerprint must be
/// roster-authorized, AND the advertised `fp` must equal the signer's
/// fingerprint (a broker may only advertise as itself). Returns `None` on
/// any failure — never panics.
pub fn fetch_advert(
    base_url: &str,
    roster: &super::roster_cache::RosterCache,
    agent: &ureq::Agent,
    node_key: &super::node_identity::NodeKey,
) -> Option<Advert> {
    let url = format!("{}/peer/advert", base_url);
    let mut rb = agent.get(&url);
    for (k, v) in node_key.sign_headers("GET", "/peer/advert", b"") {
        rb = rb.header(k.as_str(), v.as_str());
    }
    let resp = rb.call().ok()?;
    if resp.status().as_u16() != 200 {
        return None;
    }
    let node_id = resp
        .headers()
        .get("X-Node-Id")
        .and_then(|v| v.to_str().ok())?
        .to_string();
    let sig = resp
        .headers()
        .get("X-Node-Sig")
        .and_then(|v| v.to_str().ok())?
        .to_string();
    let body = resp.into_body().read_to_string().ok()?.into_bytes();
    verify_advert_body(&body, &node_id, &sig, roster)
}

/// Pure verify+parse core of [`fetch_advert`], split out for testability
/// without a live HTTP round-trip: verify `sig` over the raw `body` from
/// `node_id`, reject unless `node_id`'s fingerprint is roster-authorized AND
/// matches the advert's own claimed `fp` (self-advertisement only — a broker
/// must never be able to advertise as a different rostered peer), then parse
/// and return the `Advert`. `None` on any failure — never panics on
/// malformed/hostile JSON.
fn verify_advert_body(
    body: &[u8],
    node_id: &str,
    sig: &str,
    roster: &super::roster_cache::RosterCache,
) -> Option<Advert> {
    if !super::node_identity::verify_sig(node_id, body, sig) {
        return None;
    }
    let signer_fp = super::node_identity::fingerprint_of_pubkey_b64(node_id)?;
    if !roster.fingerprint_authorized(&signer_fp) {
        return None;
    }
    let advert: Advert = serde_json::from_slice(body).ok()?;
    if advert.fp != signer_fp {
        return None;
    }
    Some(advert)
}

/// Roster-gated, identity-bound peer acceptance.
///
/// A gossiped/scanned `(fp, url)` hint is only a HINT — `url` might be
/// controlled by an attacker who is NOT the node behind `fp` (address spoof:
/// advertise a legitimate roster member's fingerprint at an attacker-owned
/// address). `accept_peer` re-probes `url` directly and only accepts when
/// BOTH hold:
///   1. `fp` is a roster-authorized (non-revoked) node.
///   2. A live, signed `/peer/health` response fetched from `url` is signed
///      by a key whose fingerprint is exactly `fp` (not merely "some
///      roster member" — the SAME one the hint claims).
///
/// This is the production entry point; `accept_peer_with` takes the health
/// fetch as an injectable dependency so the identity-binding logic can be
/// unit-tested without a live HTTP round-trip.
///
/// `peer_key` is attached as `X-Peer-Key` on the `/peer/health` probe — that
/// endpoint is gated by `peer_handshake_auth` (shared-secret or loopback);
/// without it every non-loopback probe 401s and `accept_peer` always rejects.
pub fn accept_peer(
    fp: &str,
    url: &str,
    roster: &super::roster_cache::RosterCache,
    peer_key: &str,
) -> bool {
    accept_peer_with(fp, url, roster, |u| fetch_signed_health(u, peer_key))
}

/// Core of [`accept_peer`], with the `/peer/health` fetch injected as `health_fetch`
/// (`url -> Some((X-Node-Id, X-Node-Sig, body))`, `None` on any fetch failure).
pub fn accept_peer_with<F>(
    fp: &str,
    url: &str,
    roster: &super::roster_cache::RosterCache,
    health_fetch: F,
) -> bool
where
    F: Fn(&str) -> Option<(String, String, Vec<u8>)>,
{
    if !roster.fingerprint_authorized(fp) {
        return false;
    }
    let (node_id, sig, body) = match health_fetch(url) {
        Some(v) => v,
        None => return false,
    };
    if !super::node_identity::verify_sig(&node_id, &body, &sig) {
        return false;
    }
    match super::node_identity::fingerprint_of_pubkey_b64(&node_id) {
        Some(responder_fp) => responder_fp == fp,
        None => false,
    }
}

/// Live `GET {url}/peer/health` fetch, returning the signed identity
/// (`X-Node-Id`, `X-Node-Sig`) and raw response body for [`accept_peer_with`]
/// to verify. `None` on any network/parse failure or missing signature headers.
///
/// `peer_key` (may be empty) is attached as `X-Peer-Key` — `/peer/health` is
/// gated by `peer_handshake_auth`, which requires it for any non-loopback
/// caller (see server.rs). Routed through the mesh sidecar's CONNECT proxy
/// when `url`'s host is a WireGuard mesh address, same as `PeerClient`.
fn fetch_signed_health(url: &str, peer_key: &str) -> Option<(String, String, Vec<u8>)> {
    let health_url = format!("{}/peer/health", url);
    let host = url
        .split("://")
        .nth(1)
        .and_then(|r| r.split([':', '/']).next())
        .unwrap_or("");
    let mut cfg = ureq::Agent::config_builder().timeout_global(Some(Duration::from_secs(3)));
    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());
    let mut rb = agent.get(&health_url);
    if !peer_key.is_empty() {
        rb = rb.header("X-Peer-Key", peer_key);
    }
    let resp = rb.call().ok()?;
    if resp.status().as_u16() != 200 {
        return None;
    }
    let node_id = resp
        .headers()
        .get("X-Node-Id")
        .and_then(|v| v.to_str().ok())?
        .to_string();
    let sig = resp
        .headers()
        .get("X-Node-Sig")
        .and_then(|v| v.to_str().ok())?
        .to_string();
    let body = resp.into_body().read_to_string().ok()?.into_bytes();
    Some((node_id, sig, body))
}

/// Verify a `(fp, url)` peer claim via [`accept_peer_with`] and, only if it
/// passes, register the peer AND seed its identity cache with the fingerprint
/// that verification just proved.
///
/// Seeding here is what makes settlement work immediately. `PeerClient`'s
/// fingerprint used to be populated *only* by `check_health`, which runs every
/// 12 discovery ticks (~60s), while a peer's *workers* are stamped
/// independently from the sync payload's `source_node` — so remote workers were
/// dispatchable while their owning peer's identity cache was still empty, and
/// `executor_fp` → `get_url_for_fingerprint` resolved nothing. Every remote job
/// in that window was executed for free (executor +0, platform +0, requester
/// debited then fully refunded). The fingerprint used here is never taken on
/// trust: it must equal the fingerprint of the key that signed the peer's own
/// `/peer/health` body, and be roster-authorized.
///
/// Returns whether the peer was accepted.
fn admit_verified_peer_with<F>(
    state: &Arc<BrokerState>,
    fp: &str,
    url: &str,
    roster: &super::roster_cache::RosterCache,
    health_fetch: F,
) -> bool
where
    F: Fn(&str) -> Option<(String, String, Vec<u8>)>,
{
    if !accept_peer_with(fp, url, roster, health_fetch) {
        return false;
    }
    state.peer_manager.register_peer(url.to_string());
    state.peer_manager.set_peer_fingerprint(url, fp);
    true
}

/// One discovery round: merge signed gossip from every currently-known peer
/// (re-probing each hinted `(fp, url)` via [`admit_verified_peer_with`] before
/// trusting it), then — only when the peer set is still empty after that —
/// backfill via a mesh-subnet scan (also gated through
/// [`admit_verified_peer_with`], using the fingerprint the scanned host's own
/// signed `/peer/health` claims). Persists the resulting peer set to disk.
pub fn run_discovery_round(state: &Arc<BrokerState>, roster: &super::roster_cache::RosterCache) {
    let agent = crate::vpn::mesh_agent(Duration::from_secs(3));
    let peer_key = state.peer_manager.peer_key();

    // Merge gossip from every peer we already know about.
    for peer_url in state.peer_manager.peer_urls() {
        for (fp, url, _epoch) in fetch_gossip(&peer_url, roster, &agent, &state.node_key) {
            admit_verified_peer_with(state, &fp, &url, roster, |u| {
                fetch_signed_health(u, peer_key)
            });
        }
    }

    // Backfill via mesh scan only when we still have no peers.
    if state.peer_manager.peer_urls().is_empty() {
        if let Some(self_ip) = get_mesh_ip() {
            for url in discover_broker_peers_on_mesh_subnet(&self_ip, peer_key) {
                if let Some((node_id, sig, body)) = fetch_signed_health(&url, peer_key) {
                    if super::node_identity::verify_sig(&node_id, &body, &sig) {
                        if let Some(fp) = super::node_identity::fingerprint_of_pubkey_b64(&node_id)
                        {
                            // Reuse the health response already fetched above instead
                            // of letting `accept_peer` re-probe `/peer/health` a
                            // second time for the same host (finding 4: dedup the
                            // mesh-scan backfill's double health round-trip).
                            let fetched = (node_id.clone(), sig.clone(), body.clone());
                            admit_verified_peer_with(state, &fp, &url, roster, move |_| {
                                Some(fetched.clone())
                            });
                        }
                    }
                }
            }
        }
    }

    state.peer_manager.persist();

    refresh_adverts(state, roster);
}

/// Refresh cached price/resource adverts for every currently-known peer.
///
/// Best-effort: a failed/unreachable/unverifiable fetch simply leaves
/// whatever advert was last cached for that peer in place (staleness is
/// handled by `advert_seen_at` at selection time, not here) — a hostile or
/// down peer must never clear good cached data or panic the round.
pub(crate) fn refresh_adverts(state: &Arc<BrokerState>, roster: &super::roster_cache::RosterCache) {
    let agent = crate::vpn::mesh_agent(Duration::from_secs(3));
    for peer_url in state.peer_manager.peer_urls() {
        if let Some(advert) = fetch_advert(&peer_url, roster, &agent, &state.node_key) {
            state.peer_manager.set_peer_advert(&peer_url, advert);
        }
    }
}

/// Result from probing a worker
#[derive(Debug, Clone, serde::Deserialize)]
struct WorkerProbeResult {
    name: Option<String>,
    worker_type: Option<String>,
    resources: Option<WorkerResources>,
    pricing: Option<WorkerPricing>,
    tags: Option<Vec<String>>,
    /// Hardware details reported by worker
    #[serde(default)]
    hardware: Option<HardwareInfo>,
}

/// Trait for colored output
trait ColorExt {
    fn cyan(&self) -> String;
    fn yellow(&self) -> String;
    fn green(&self) -> String;
}

impl ColorExt for &str {
    fn cyan(&self) -> String {
        format!("\x1b[36m{}\x1b[0m", self)
    }
    fn yellow(&self) -> String {
        format!("\x1b[33m{}\x1b[0m", self)
    }
    fn green(&self) -> String {
        format!("\x1b[32m{}\x1b[0m", self)
    }
}

/// Is `host` this machine's loopback interface? Local-worker auto-discovery
/// is gated on this: a broker owns and manages ONLY its own local workers —
/// anything reached over a non-loopback address (mesh subnet IPs, peer
/// hostnames, LAN IPs) is another node's concern, never adopted as local.
/// Parse `ZAKURO_WORKERS` into (host, port) pairs.
///
/// Accepts comma-separated `host:port` or bare `host` (which takes
/// `default_port`). Blank entries are ignored.
pub(crate) fn parse_local_workers(raw: &str, default_port: u16) -> Vec<(String, u16)> {
    raw.split(',')
        .map(|s| s.trim())
        .filter(|s| !s.is_empty())
        .filter_map(|entry| match entry.rsplit_once(':') {
            Some((host, port)) => {
                let host = host.trim();
                match port.trim().parse::<u16>() {
                    Ok(p) if !host.is_empty() => Some((host.to_string(), p)),
                    _ => None,
                }
            }
            None => Some((entry.to_string(), default_port)),
        })
        .collect()
}

/// True when `host:port` was explicitly named as one of THIS broker's own
/// workers via `ZAKURO_WORKERS`.
pub(crate) fn is_explicit_local_worker(
    local_workers: &[(String, u16)],
    host: &str,
    port: u16,
) -> bool {
    local_workers.iter().any(|(h, p)| h == host && *p == port)
}

fn is_loopback_host(host: &str) -> bool {
    host == "127.0.0.1" || host == "localhost" || host == "::1"
}

/// Detect the appropriate discovery mode based on network availability
/// Bare node name from a `zc://node-…` handle (or pass-through for a bare name).
pub(crate) fn strip_zc_node(s: &str) -> &str {
    s.strip_prefix("zc://").unwrap_or(s)
}

/// Fingerprint to stamp onto a peer-synced worker's `node_fp`, derived from its
/// `source_node` (`node-{fingerprint}`, per [`strip_zc_node`]). `None` if
/// `source_node` is absent or would reduce to an empty fingerprint.
/// Dedup predicate for peer-synced workers: does `existing` represent the SAME
/// key-derived identity as the incoming peer worker (`name`/`uri` at source
/// node `source_node`, e.g. `node-i9`)?
///
/// Parse the provider fields off one /peer/workers entry. Absent or
/// malformed fields fall back to the same defaults a plain execute worker
/// registers with (specialized, no models, unpriced), so a peer running an
/// older zc that does not emit them syncs exactly as before — the worker
/// just is not indexed as a model provider (zc#172).
fn provider_fields_from_json(
    w: &serde_json::Value,
) -> (crate::broker::worker::ProviderType, Vec<String>, f64) {
    let provider_type = w
        .get("provider_type")
        .cloned()
        .and_then(|v| serde_json::from_value(v).ok())
        .unwrap_or_default();
    let served_models = w
        .get("served_models")
        .and_then(|v| v.as_array())
        .map(|a| {
            a.iter()
                .filter_map(|m| m.as_str())
                .map(|m| m.to_string())
                .collect()
        })
        .unwrap_or_default();
    let price_per_mtok = w
        .get("price_per_mtok")
        .and_then(|v| v.as_f64())
        .unwrap_or(0.0);
    (provider_type, served_models, price_per_mtok)
}

/// Matches only when `existing` is itself a peer worker (`source_node: Some`)
/// from the SAME source node fingerprint, and additionally shares the name or
/// uri. A local worker (`existing.source_node == None`) never matches — no
/// amount of name/uri overlap can make a local worker "the same" as a peer
/// worker. Two peer workers from DIFFERENT nodes with the same bare name
/// (e.g. both "worker-lxd" because every container's hostname is "lxd") are
/// also never conflated, since their fingerprints differ.
fn is_same_peer_worker(
    existing: &super::worker::Worker,
    name: &str,
    uri: &str,
    source_node: &str,
) -> bool {
    match existing.source_node.as_deref() {
        Some(existing_sn) => {
            let existing_fp = existing_sn.strip_prefix("node-").unwrap_or(existing_sn);
            let incoming_fp = source_node.strip_prefix("node-").unwrap_or(source_node);
            existing_fp == incoming_fp && (existing.name == name || existing.uri == uri)
        }
        None => false,
    }
}

/// Merge `PeerManager`-known broker URLs (dynamically discovered via
/// mesh-subnet/localhost scan or gossip admission) into the `(host, port)`
/// list `scan_peers` should additionally fetch `/peer/workers` from this
/// tick.
///
/// `dynamic_urls` are full `"http://host:port"` broker base URLs as returned
/// by `PeerManager::peer_urls()` — PeerManager already excludes this
/// broker's own address (see `PeerManager::new_with_node_key`'s `own_addr`
/// exclusion), so no additional self-filtering happens here. `explicit_hosts`
/// is the `(host, broker_port)` set `scan_peers` already visits from
/// `DiscoveryConfig.peers` (`ZAKURO_PEERS`) — passed in so a broker present
/// in both lists (e.g. gossip re-admits one the operator also named
/// explicitly) is scanned once per tick, not twice. A malformed URL (should
/// never come from `PeerManager`, but this function must not trust that from
/// the outside) is skipped rather than panicking or poisoning the scan.
pub(crate) fn resolve_dynamic_peer_broker_addrs(
    dynamic_urls: &[String],
    explicit_hosts: &std::collections::HashSet<(String, u16)>,
) -> Vec<(String, u16)> {
    let mut seen: std::collections::HashSet<(String, u16)> = std::collections::HashSet::new();
    let mut out = Vec::new();

    for url in dynamic_urls {
        let rest = url
            .strip_prefix("http://")
            .or_else(|| url.strip_prefix("https://"))
            .unwrap_or(url);
        let rest = rest.trim_end_matches('/');
        let (host, port_str) = match rest.rsplit_once(':') {
            Some(v) => v,
            None => continue,
        };
        let port: u16 = match port_str.parse() {
            Ok(p) => p,
            Err(_) => continue,
        };
        if host.is_empty() {
            continue;
        }

        let key = (host.to_string(), port);
        if explicit_hosts.contains(&key) || !seen.insert(key.clone()) {
            continue;
        }
        out.push(key);
    }

    out
}

/// Fingerprint to stamp onto a peer-synced worker's `node_fp`, derived from its
/// `source_node` (`node-{fingerprint}`, per [`strip_zc_node`]). `None` if
/// `source_node` is absent or would reduce to an empty fingerprint.
fn peer_node_fp(worker: &super::worker::Worker) -> Option<String> {
    let sn = worker.source_node.as_deref()?;
    let fp = sn.strip_prefix("node-").unwrap_or(sn);
    if fp.is_empty() {
        None
    } else {
        Some(fp.to_string())
    }
}

pub fn detect_discovery_mode(preferred_subnet: &str) -> DiscoveryMode {
    // First, check for WireGuard IP (env var or interface detection)
    if let Some(wireguard_ip) = get_mesh_ip() {
        // Extract subnet from WireGuard IP (e.g., "100.64.0.5" -> "100.64.0")
        let parts: Vec<&str> = wireguard_ip.split('.').collect();
        if parts.len() == 4 {
            let subnet = format!("{}.{}.{}", parts[0], parts[1], parts[2]);
            return DiscoveryMode::WireGuard { subnet };
        }
    }

    // Check if preferred subnet is reachable (might be on VPN/custom network)
    let test_ip = format!("{}.1", preferred_subnet);
    if let Ok(addr) = format!("{}:1", test_ip).parse() {
        if TcpStream::connect_timeout(&addr, Duration::from_millis(50)).is_ok() {
            return DiscoveryMode::WireGuard {
                subnet: preferred_subnet.to_string(),
            };
        }
    }

    // If peers are configured, use WireGuard mode with a dummy subnet
    // (actual discovery happens via peer probing, not subnet scan)
    if std::env::var("ZAKURO_PEERS")
        .map(|v| !v.is_empty())
        .unwrap_or(false)
    {
        return DiscoveryMode::WireGuard {
            subnet: "peers".to_string(),
        };
    }

    // Fall back to local mode
    DiscoveryMode::Local
}

/// Get the best available IP for this node to advertise to peers.
/// Prefers the WireGuard mesh IP (10.13.13.0/24) over the primary LAN IP.
/// Returns None only if no non-loopback IP can be determined.
pub fn get_effective_node_ip() -> Option<String> {
    if let Some(ip) = get_mesh_ip() {
        return Some(ip);
    }
    // Fall back to the primary LAN IP (for brokers not yet on the mesh)
    if let Ok(socket) = std::net::UdpSocket::bind("0.0.0.0:0") {
        if socket.connect("8.8.8.8:80").is_ok() {
            if let Ok(addr) = socket.local_addr() {
                let ip = addr.ip().to_string();
                if ip != "127.0.0.1" && ip != "::1" {
                    return Some(ip);
                }
            }
        }
    }
    None
}

/// Discover other broker instances on localhost by probing /peer/health.
/// Used when ZAKURO_PEERS is empty so brokers on the same machine can find each other
/// without manual config. Excludes `self_port` from the result.
/// Returns URLs like "http://127.0.0.1:9001".
pub fn discover_broker_peers_on_localhost(
    self_port: u16,
    port_start: u16,
    port_end: u16,
) -> Vec<String> {
    let mut out = Vec::new();
    let agent = ureq::Agent::new_with_config(
        ureq::Agent::config_builder()
            .timeout_global(Some(Duration::from_millis(500)))
            .build(),
    );
    for port in port_start..=port_end {
        if port == self_port {
            continue;
        }
        let url = format!("http://127.0.0.1:{}/peer/health", port);
        if agent
            .get(&url)
            .call()
            .map(|r| r.status().as_u16() == 200)
            .unwrap_or(false)
        {
            out.push(format!("http://127.0.0.1:{}", port));
        }
    }
    out
}

/// Discover peer brokers on the WireGuard mesh subnet (`10.13.13.0/24`) by
/// probing `/peer/health` on port 9000 for every host in the subnet.
///
/// Best-effort and bounded: run only when P2P is enabled, `ZAKURO_PEERS` is
/// empty, and this node already has a mesh IP (see `get_mesh_ip`) — callers
/// are responsible for that gating, this function just does the scan.
/// Concurrency is capped in batches (mirrors the parallel-probe pattern used
/// by `Discovery::scan_localhost`) so we never open 254 sockets at once.
/// Any error or timeout for a given host is swallowed silently; a partial
/// or empty result is a normal outcome, not a failure.
/// `peer_key` (may be empty) is attached as `X-Peer-Key` on every probe —
/// `/peer/health` is gated by `peer_handshake_auth`, which requires it for
/// any non-loopback caller (mesh hosts always are); without it every probe
/// 401s and this scan always returns empty.
pub fn discover_broker_peers_on_mesh_subnet(self_ip: &str, peer_key: &str) -> Vec<String> {
    const PORT: u16 = 9000;
    const BATCH_SIZE: usize = 32;
    const TIMEOUT: Duration = Duration::from_millis(400);

    let hosts: Vec<String> = (1u8..=254)
        .map(|h| format!("10.13.13.{}", h))
        .filter(|ip| ip != self_ip)
        .collect();

    let mut out = Vec::new();
    for batch in hosts.chunks(BATCH_SIZE) {
        let (tx, rx) = std::sync::mpsc::channel();
        let mut handles = Vec::with_capacity(batch.len());
        for host in batch {
            let host = host.clone();
            let tx = tx.clone();
            let peer_key = peer_key.to_string();
            handles.push(thread::spawn(move || {
                let agent = ureq::Agent::new_with_config(
                    ureq::Agent::config_builder()
                        .timeout_global(Some(TIMEOUT))
                        .build(),
                );
                let url = format!("http://{}:{}/peer/health", host, PORT);
                let mut rb = agent.get(&url);
                if !peer_key.is_empty() {
                    rb = rb.header("X-Peer-Key", &peer_key);
                }
                let ok = rb
                    .call()
                    .map(|r| r.status().as_u16() == 200)
                    .unwrap_or(false);
                if ok {
                    let _ = tx.send(format!("http://{}:{}", host, PORT));
                }
            }));
        }
        drop(tx);
        out.extend(rx.iter());
        for h in handles {
            let _ = h.join();
        }
    }
    out
}

/// Get this node's WireGuard mesh IP (Zakuro mesh, `10.13.13.0/24`).
///
/// WireGuard is decommissioned; the mesh now runs on WireGuard. Detection:
/// `ZAKURO_MESH_IP` env (or legacy `ZAKURO_WIREGUARD_IP`) first, then a
/// `zakuro0`/`wg*` interface carrying a `10.13.13.x` address (present when the
/// broker shares the WireGuard vpn container's netns).
pub fn get_mesh_ip() -> Option<String> {
    for var in ["ZAKURO_MESH_IP", "ZAKURO_WIREGUARD_IP"] {
        if let Ok(ip) = std::env::var(var) {
            if !ip.is_empty() {
                return Some(ip);
            }
        }
    }

    #[cfg(unix)]
    {
        for iface in ifaces::Interface::get_all().ok()?.into_iter() {
            // WireGuard mesh interface — zc's vpn module names it `zakuro0`; accept
            // any `wg*` too. The mesh subnet is 10.13.13.0/24.
            if iface.name == "zakuro0" || iface.name.starts_with("wg") {
                if let Some(addr) = iface.addr {
                    let addr_str = addr.to_string();
                    let ip = addr_str.trim_end_matches(":0");
                    if ip.starts_with("10.13.13.") {
                        return Some(ip.to_string());
                    }
                }
            }
        }
    }
    None
}

#[cfg(test)]
mod kdi_registry_dedup_tests {
    use super::is_same_peer_worker;
    use crate::broker::worker::{Worker, WorkerRegistration, WorkerRegistry};

    fn peer_worker(id: &str, name: &str, uri: &str, source_node: &str) -> Worker {
        let mut w = Worker::new(id.into(), name.into(), uri.into(), "zakuro".into());
        w.source_node = Some(source_node.into());
        w
    }

    /// Unit-test the dedup predicate directly: local vs peer, same name,
    /// different fp → not equal.
    #[test]
    fn predicate_local_worker_never_matches_a_peer() {
        let local = Worker::new(
            "local1".into(),
            "worker-lxd".into(),
            "http://127.0.0.1:3960".into(),
            "zakuro".into(),
        );
        assert!(local.source_node.is_none());
        assert!(!is_same_peer_worker(
            &local,
            "worker-lxd",
            "http://10.13.13.9:3960",
            "node-cc1fingerprint"
        ));
    }

    #[test]
    fn predicate_same_name_different_source_node_not_equal() {
        let peer_a = peer_worker(
            "a1",
            "worker-lxd",
            "http://10.13.13.9:3960",
            "node-i9fingerprint",
        );
        assert!(!is_same_peer_worker(
            &peer_a,
            "worker-lxd",
            "http://10.13.13.10:3960",
            "node-cc1fingerprint",
        ));
    }

    #[test]
    fn predicate_same_name_same_source_node_is_equal() {
        let peer_a = peer_worker(
            "a1",
            "worker-lxd",
            "http://10.13.13.9:3960",
            "node-i9fingerprint",
        );
        assert!(is_same_peer_worker(
            &peer_a,
            "worker-lxd",
            "http://10.13.13.9:3960",
            "node-i9fingerprint",
        ));
    }

    /// Two brokers with the SAME hostname ("lxd") → same worker name
    /// "worker-lxd" but DIFFERENT node fingerprints. Simulate the registry
    /// side of peer-sync directly (bypassing HTTP): both workers must land
    /// as distinct registry entries, each with the correct source_node and
    /// a distinct zc_uri(); a broker's own local worker-lxd must not be
    /// overwritten/hidden by either.
    #[test]
    fn same_hostname_different_node_fp_yields_distinct_registry_entries() {
        let registry = WorkerRegistry::new();

        // This broker's own local worker (source_node: None).
        let local = registry.register(WorkerRegistration {
            name: "worker-lxd".into(),
            uri: "http://127.0.0.1:3960".into(),
            worker_type: "zakuro".into(),
            resources: Default::default(),
            pricing: Default::default(),
            tags: vec![],
            max_timeout_secs: 0.0,
            hardware: Default::default(),
            wireguard_ip: None,
            is_docker: None,
            source_node: None,
            explicit_local: false,
            provider_type: Default::default(),
            served_models: vec![],
            price_per_mtok: 0.0,
        });
        let local = registry.set_node_fp(&local.id, "cc1fingerprint").unwrap();

        // Peer worker synced from node "i9", also named "worker-lxd".
        let peer_i9 = registry.register(WorkerRegistration {
            name: "worker-lxd".into(),
            uri: "http://10.13.13.9:3960".into(),
            worker_type: "zakuro".into(),
            resources: Default::default(),
            pricing: Default::default(),
            tags: vec![],
            max_timeout_secs: 0.0,
            hardware: Default::default(),
            wireguard_ip: Some("10.13.13.9".into()),
            is_docker: None,
            source_node: Some("node-i9fingerprint".into()),
            explicit_local: false,
            provider_type: Default::default(),
            served_models: vec![],
            price_per_mtok: 0.0,
        });
        let peer_i9 = registry.set_node_fp(&peer_i9.id, "i9fingerprint").unwrap();

        // Peer worker synced from node "msi", also named "worker-lxd".
        let peer_msi = registry.register(WorkerRegistration {
            name: "worker-lxd".into(),
            uri: "http://10.13.13.11:3960".into(),
            worker_type: "zakuro".into(),
            resources: Default::default(),
            pricing: Default::default(),
            tags: vec![],
            max_timeout_secs: 0.0,
            hardware: Default::default(),
            wireguard_ip: Some("10.13.13.11".into()),
            is_docker: None,
            source_node: Some("node-msifingerprint".into()),
            explicit_local: false,
            provider_type: Default::default(),
            served_models: vec![],
            price_per_mtok: 0.0,
        });
        let peer_msi = registry
            .set_node_fp(&peer_msi.id, "msifingerprint")
            .unwrap();

        // Dedup predicate: the local worker is never "already registered" as
        // either peer, and the two peers are never conflated with each other.
        assert!(!is_same_peer_worker(
            &local,
            "worker-lxd",
            "http://10.13.13.9:3960",
            "node-i9fingerprint"
        ));
        assert!(!is_same_peer_worker(
            &local,
            "worker-lxd",
            "http://10.13.13.11:3960",
            "node-msifingerprint"
        ));
        assert!(!is_same_peer_worker(
            &peer_i9,
            "worker-lxd",
            "http://10.13.13.11:3960",
            "node-msifingerprint"
        ));
        // But a repeat sync of the SAME peer worker does match (refresh path).
        assert!(is_same_peer_worker(
            &peer_i9,
            "worker-lxd",
            "http://10.13.13.9:3960",
            "node-i9fingerprint"
        ));

        // All three are distinct registry entries.
        let all = registry.list();
        assert_eq!(
            all.len(),
            3,
            "local + two same-named peers must all persist"
        );

        // Each carries the correct source_node.
        assert!(local.source_node.is_none());
        assert_eq!(peer_i9.source_node.as_deref(), Some("node-i9fingerprint"));
        assert_eq!(peer_msi.source_node.as_deref(), Some("node-msifingerprint"));

        // Distinct key-derived zc_uri()s despite the identical bare name.
        let uri_local = local.zc_uri();
        let uri_i9 = peer_i9.zc_uri();
        let uri_msi = peer_msi.zc_uri();
        assert_ne!(uri_local, uri_i9);
        assert_ne!(uri_local, uri_msi);
        assert_ne!(uri_i9, uri_msi);
        assert!(uri_i9.starts_with("zc://worker-i9fingerprint-"));
        assert!(uri_msi.starts_with("zc://worker-msifingerprint-"));
        assert!(uri_local.starts_with("zc://worker-cc1fingerprint-"));
    }
}

#[cfg(test)]
mod price_sanitize_tests {
    use super::sanitize_price;

    #[test]
    fn strip_zc_node_handles_both_forms() {
        assert_eq!(super::strip_zc_node("zc://node-i9"), "node-i9");
        assert_eq!(super::strip_zc_node("node-i9"), "node-i9");
    }

    #[test]
    fn peer_synced_worker_node_fp_survives_round_trip() {
        // Simulate what fetch_workers_from_peer_broker does: a worker registered
        // locally with source_node carrying the source broker's `node-{fp}` handle
        // (as read off the /peer/workers JSON `node` field, via strip_zc_node).
        use crate::broker::worker::Worker;
        let mut w = Worker::new(
            "id1".into(),
            "worker-x".into(),
            "http://10.13.13.9:3960".into(),
            "zakuro".into(),
        );
        w.source_node = Some("node-deadbeefcafebabe".into());
        w.slot = "3960".into();

        let fp = super::peer_node_fp(&w).expect("fingerprint recovered from source_node");
        assert!(!fp.is_empty());
        w.node_fp = fp;

        // Client-facing uri must be well-formed: fp present, no double-dash.
        let uri = w.zc_uri();
        assert!(uri.starts_with("zc://worker-deadbeefcafebabe-"));
        assert!(!uri.contains("--"), "malformed uri (empty node_fp): {uri}");
    }

    #[test]
    fn peer_node_fp_none_without_source_node() {
        use crate::broker::worker::Worker;
        let w = Worker::new(
            "id2".into(),
            "worker-y".into(),
            "http://127.0.0.1:3960".into(),
            "zakuro".into(),
        );
        assert!(super::peer_node_fp(&w).is_none());
    }

    #[test]
    fn sanitize_price_accepts_valid() {
        assert!((sanitize_price(Some(2.5), 3.6, 1000.0) - 2.5).abs() < 1e-9);
        assert_eq!(sanitize_price(Some(0.0), 3.6, 1000.0), 0.0);
    }

    #[test]
    fn sanitize_price_rejects_bad_values() {
        assert_eq!(sanitize_price(Some(-1.0), 3.6, 1000.0), 3.6); // negative
        assert_eq!(sanitize_price(Some(f64::NAN), 3.6, 1000.0), 3.6); // NaN
        assert_eq!(sanitize_price(Some(f64::INFINITY), 3.6, 1000.0), 3.6); // inf
        assert_eq!(sanitize_price(Some(5000.0), 3.6, 1000.0), 3.6); // over max
        assert_eq!(sanitize_price(None, 3.6, 1000.0), 3.6); // missing
    }
}

/// CHANGE 1 tests: worker auto-discovery is constrained to localhost — a
/// broker owns and manages ONLY its own local workers; a reachable worker on
/// a non-loopback mesh address must never be adopted as local (source_node:
/// None).
#[cfg(test)]
mod localhost_only_discovery_tests {
    use super::*;
    use crate::broker::BrokerState;
    use std::io::Read;
    use std::net::TcpListener;
    use std::sync::atomic::{AtomicU16, Ordering};

    #[test]
    fn explicit_local_workers_are_parsed_from_the_env_list() {
        // "host:port" and bare "host" (which takes worker_port).
        let parsed = parse_local_workers("172.17.0.1:3960, 10.88.0.1 ,", 3960);
        assert_eq!(
            parsed,
            vec![
                ("172.17.0.1".to_string(), 3960u16),
                ("10.88.0.1".to_string(), 3960u16)
            ]
        );
        assert!(parse_local_workers("", 3960).is_empty());
    }

    #[test]
    fn explicit_local_workers_bypass_the_loopback_gate() {
        // Regression (staging 2026-08-14): a broker started with
        // `network_mode: container:<wireguard sidecar>` sees its OWN worker on
        // the host network at the container gateway (172.17.0.1:3960), never
        // on loopback. `is_loopback_host` rejected it, /workers stayed empty
        // and every job returned HTTP 503. An operator naming the address
        // explicitly is asserting "this worker is mine", so the gate must
        // yield -- while still refusing anything NOT named.
        assert!(!is_loopback_host("172.17.0.1"));
        let explicit = parse_local_workers("172.17.0.1:3960", 3960);
        assert!(is_explicit_local_worker(&explicit, "172.17.0.1", 3960));
        // A different host/port is still not adopted.
        assert!(!is_explicit_local_worker(&explicit, "10.13.13.9", 3960));
        assert!(!is_explicit_local_worker(&explicit, "172.17.0.1", 3961));
    }

    #[test]
    fn is_loopback_host_gate() {
        assert!(is_loopback_host("127.0.0.1"));
        assert!(is_loopback_host("localhost"));
        assert!(is_loopback_host("::1"));
        // Same loopback /8, but not the canonical local address our broker
        // binds to — treated as a remote mesh host, never "local".
        assert!(!is_loopback_host("127.0.0.2"));
        assert!(!is_loopback_host("10.13.13.9"));
        assert!(!is_loopback_host("100.64.0.5"));
    }

    /// Reserve a port on `bind_ip` by binding it and **keeping the listener**.
    ///
    /// The previous helper returned a bare `u16` after dropping its probe
    /// listener, so the port was unowned between the probe and
    /// `spawn_fake_worker`'s re-bind — a TOCTOU window in which a concurrent
    /// test thread could take it, killing this test with
    /// `AddrInUse (os error 98)`. Handing the live listener straight to
    /// `spawn_fake_worker` means the port is never released, so the window does
    /// not exist.
    ///
    /// `bind_ip` matters: callers serve on addresses other than `127.0.0.1`
    /// (e.g. `127.0.0.2`, deliberately non-canonical so the worker is treated as
    /// a remote mesh host). A port proven free on loopback is not necessarily
    /// free on another interface, so the reservation is taken on the SAME
    /// address the worker will serve on.
    ///
    /// The counter's base sits well clear of `integration_tests::free_port`'s
    /// 20000 band, which is a separate static in a separate module.
    fn reserve_port(bind_ip: &str) -> TcpListener {
        static NEXT: AtomicU16 = AtomicU16::new(41000);
        loop {
            let port = NEXT.fetch_add(1, Ordering::Relaxed);
            assert!(port < 50000, "reserve_port exhausted the test port range");
            if let Ok(listener) = TcpListener::bind((bind_ip, port)) {
                return listener;
            }
        }
    }

    /// Spawn a minimal fake worker on an already-reserved `listener` that answers
    /// /health and /info with just enough JSON for `probe_worker` to accept it.
    /// Takes the bound listener (not a port number) so no port is ever released
    /// and re-bound — see [`reserve_port`].
    fn spawn_fake_worker(listener: TcpListener, name: &str) {
        let name = name.to_string();
        std::thread::spawn(move || {
            for stream in listener.incoming() {
                let mut stream = match stream {
                    Ok(s) => s,
                    Err(_) => continue,
                };
                let mut buf = [0u8; 1024];
                let n = stream.read(&mut buf).unwrap_or(0);
                let req = String::from_utf8_lossy(&buf[..n]);
                let path = req.split_whitespace().nth(1).unwrap_or("/");
                let body = if path.starts_with("/info") {
                    format!(
                        r#"{{"name":"{}","worker_type":"zakuro","resources":{{"cpus_available":1.0,"cpus_total":1.0,"memory_available":1,"memory_total":1,"gpus_available":0,"gpus_total":0}}}}"#,
                        name
                    )
                } else {
                    "{}".to_string()
                };
                let resp = format!(
                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                    body.len(),
                    body
                );
                use std::io::Write;
                let _ = stream.write_all(resp.as_bytes());
            }
        });
    }

    #[test]
    fn only_loopback_worker_is_registered_locally() {
        // Reserve on the exact address each fake worker will serve on, and hand
        // the live listener over without ever unbinding it.
        let local_listener = reserve_port("127.0.0.1");
        let remote_listener = reserve_port("127.0.0.2");
        let local_port = local_listener.local_addr().unwrap().port();
        let remote_port = remote_listener.local_addr().unwrap().port();

        spawn_fake_worker(local_listener, "worker-local");
        // "Remote" mesh worker: reachable (loopback /8 covers 127.0.0.2 on
        // Linux), but not the canonical "127.0.0.1" — must be rejected by
        // the loopback gate before any registration happens.
        spawn_fake_worker(remote_listener, "worker-remote");
        std::thread::sleep(Duration::from_millis(50));

        let state = Arc::new(BrokerState::new());
        let config = DiscoveryConfig {
            subnet: "10.13.13".to_string(),
            worker_port: local_port,
            extra_ports: vec![],
            scan_port_range: None,
            interval_secs: 9999,
            enable_scan: false,
            enable_dns: false,
            peers: vec![],
            local_workers: vec![],
        };
        let discovery = Discovery {
            config,
            state: Arc::clone(&state),
            mode: DiscoveryMode::Local,
        };

        // Local probe — must register.
        discovery.try_register_worker("127.0.0.1", local_port, false);
        // Non-loopback probe — must be a no-op even though the worker is
        // live and reachable.
        discovery.try_register_worker("127.0.0.2", remote_port, false);

        let workers = state.workers.list();
        assert_eq!(
            workers.len(),
            1,
            "only the loopback worker should be registered: {:?}",
            workers.iter().map(|w| &w.uri).collect::<Vec<_>>()
        );
        assert_eq!(workers[0].name, "worker-local");
        assert!(
            workers[0].source_node.is_none(),
            "the sole registered worker must be local (source_node: None)"
        );
        assert!(workers[0].uri.contains("127.0.0.1"));
    }
}

/// Task 3 tests: signed `GET /peer/peers` gossip client.
#[cfg(test)]
mod gossip_tests {
    use super::verify_gossip_body;
    use crate::broker::node_identity::NodeKey;
    use crate::broker::roster_cache::RosterCache;

    #[test]
    fn fetch_gossip_rejects_unrostered_signer() {
        // Body signed by key K; roster does NOT contain K -> empty.
        let key = NodeKey::generate();
        let body = br#"{"peers":[{"fp":"abc123","url":"http://10.13.13.9:9000","epoch":42}]}"#;
        let sig = key.sign(body);

        let roster = RosterCache::from_entries(vec![]); // empty roster
        let peers = verify_gossip_body(body, &key.public_b64(), &sig, &roster);
        assert!(peers.is_empty());
    }

    #[test]
    fn fetch_gossip_accepts_rostered_signer() {
        let key = NodeKey::generate();
        let body = br#"{"peers":[{"fp":"abc123","url":"http://10.13.13.9:9000","epoch":42}]}"#;
        let sig = key.sign(body);

        let roster = RosterCache::from_entries(vec![(key.public_b64(), false)]);
        let peers = verify_gossip_body(body, &key.public_b64(), &sig, &roster);
        assert_eq!(
            peers,
            vec![(
                "abc123".to_string(),
                "http://10.13.13.9:9000".to_string(),
                42u64
            )]
        );
    }

    #[test]
    fn fetch_gossip_rejects_tampered_body() {
        let key = NodeKey::generate();
        let body = br#"{"peers":[{"fp":"abc123","url":"http://10.13.13.9:9000","epoch":42}]}"#;
        let sig = key.sign(body);
        let roster = RosterCache::from_entries(vec![(key.public_b64(), false)]);

        let tampered = br#"{"peers":[{"fp":"evil","url":"http://10.13.13.99:9000","epoch":1}]}"#;
        let peers = verify_gossip_body(tampered, &key.public_b64(), &sig, &roster);
        assert!(peers.is_empty());
    }

    #[test]
    fn fetch_gossip_rejects_revoked_signer() {
        let key = NodeKey::generate();
        let body = br#"{"peers":[{"fp":"abc123","url":"http://10.13.13.9:9000","epoch":42}]}"#;
        let sig = key.sign(body);

        let roster = RosterCache::from_entries(vec![(key.public_b64(), true)]); // revoked
        let peers = verify_gossip_body(body, &key.public_b64(), &sig, &roster);
        assert!(peers.is_empty());
    }

    #[test]
    fn fetch_gossip_returns_empty_on_unreachable_peer() {
        // No live server on this port — network call must fail closed, not panic.
        let roster = RosterCache::from_entries(vec![]);
        let agent = ureq::Agent::new_with_config(
            ureq::Agent::config_builder()
                .timeout_global(Some(std::time::Duration::from_millis(200)))
                .build(),
        );
        let key = crate::broker::node_identity::NodeKey::generate();
        let peers = super::fetch_gossip("http://127.0.0.1:1", &roster, &agent, &key);
        assert!(peers.is_empty());
    }

    /// Finding 1 (wiring): `fetch_gossip` must sign its outbound `GET
    /// /peer/peers` the same way the real server's `check_node_sig` gate
    /// verifies it — a plain unauthenticated request 401s. This spins a
    /// minimal server that replicates that gate (`node_identity::verify_request`
    /// against a rostered signer) and drives it end-to-end through
    /// `fetch_gossip`, proving the loop's request now actually passes.
    #[test]
    fn fetch_gossip_signs_request_so_check_node_sig_gate_accepts_it() {
        use crate::broker::node_identity::{verify_request, ReplayGuard};

        let responder_key = NodeKey::generate();
        let responder_pub = responder_key.public_b64();
        let caller_key = NodeKey::generate();
        let caller_pub = caller_key.public_b64();

        let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
        let port = server.server_addr().to_ip().unwrap().port();
        std::thread::spawn(move || {
            let guard = ReplayGuard::new();
            for req in server.incoming_requests() {
                let headers: Vec<(String, String)> = req
                    .headers()
                    .iter()
                    .map(|h| (h.field.to_string(), h.value.to_string()))
                    .collect();
                let get = |name: &str| -> Option<String> {
                    headers
                        .iter()
                        .find(|(k, _)| k.eq_ignore_ascii_case(name))
                        .map(|(_, v)| v.clone())
                };
                let is_rostered = |id: &str| id == caller_pub;
                let now = crate::broker::node_identity::now_secs();
                // Same gate shape as server.rs's check_node_sig/handle_peer_peers:
                // strict — no permissive fallback for /peer/peers.
                let verified =
                    verify_request(&is_rostered, &guard, "GET", "/peer/peers", b"", &get, now);

                let resp = match verified {
                    Ok(_) => {
                        let body = br#"{"peers":[{"fp":"abc123","url":"http://10.13.13.9:9000","epoch":1}]}"#.to_vec();
                        let sig = responder_key.sign(&body);
                        tiny_http::Response::from_data(body)
                            .with_status_code(200)
                            .with_header(
                                tiny_http::Header::from_bytes(
                                    "X-Node-Id",
                                    responder_key.public_b64(),
                                )
                                .unwrap(),
                            )
                            .with_header(tiny_http::Header::from_bytes("X-Node-Sig", sig).unwrap())
                    }
                    Err(_) => {
                        tiny_http::Response::from_string("unauthorized").with_status_code(401)
                    }
                };
                let _ = req.respond(resp);
            }
        });

        let base_url = format!("http://127.0.0.1:{}", port);
        let roster = RosterCache::from_entries(vec![(responder_pub, false)]);
        let agent = ureq::Agent::new_with_config(
            ureq::Agent::config_builder()
                .timeout_global(Some(std::time::Duration::from_secs(3)))
                .build(),
        );

        // Signed with the rostered caller key → server's gate accepts, gossip parses.
        let peers = super::fetch_gossip(&base_url, &roster, &agent, &caller_key);
        assert_eq!(
            peers,
            vec![(
                "abc123".to_string(),
                "http://10.13.13.9:9000".to_string(),
                1u64
            )],
            "signed request must pass the server's check_node_sig gate and return gossip"
        );

        // An unrostered signer is rejected server-side → 401 → empty (fails closed).
        let unrostered_key = NodeKey::generate();
        let peers2 = super::fetch_gossip(&base_url, &roster, &agent, &unrostered_key);
        assert!(
            peers2.is_empty(),
            "an unrostered signer must be rejected by the server's gate"
        );
    }
}

/// Task 4 tests: roster-gated, identity-bound peer acceptance.
#[cfg(test)]
mod accept_peer_tests {
    use super::accept_peer_with;
    use crate::broker::node_identity::NodeKey;
    use crate::broker::roster_cache::RosterCache;

    /// Stub health-fetch: always answers with a health body signed by `key`.
    fn signed_health_ok(key: &NodeKey) -> impl Fn(&str) -> Option<(String, String, Vec<u8>)> + '_ {
        move |_url| {
            let body = b"{\"status\":\"healthy\"}".to_vec();
            let sig = key.sign(&body);
            Some((key.public_b64(), sig, body))
        }
    }

    /// Stub health-fetch: answers with a health body signed by a DIFFERENT key
    /// than the one the roster/claim expects — simulates an address-spoofed
    /// peer (attacker owns `url`, but not the roster-authorized `fp`).
    fn signed_health_by_other() -> impl Fn(&str) -> Option<(String, String, Vec<u8>)> {
        let other = NodeKey::generate();
        move |_url| {
            let body = b"{\"status\":\"healthy\"}".to_vec();
            let sig = other.sign(&body);
            Some((other.public_b64(), sig, body))
        }
    }

    #[test]
    fn accept_peer_requires_roster_and_matching_signed_health() {
        let key = NodeKey::generate();
        let fp = key.fingerprint();
        let roster = RosterCache::from_entries(vec![(key.public_b64(), false)]);

        // fp is roster-authorized AND the fetched health is signed by that same key -> accepted.
        assert!(accept_peer_with(
            &fp,
            "http://10.13.13.17:9000",
            &roster,
            signed_health_ok(&key)
        ));

        // Same url/fp claim, but health is signed by a DIFFERENT key -> rejected
        // (address spoof: url does not actually belong to `fp`).
        assert!(!accept_peer_with(
            &fp,
            "http://10.13.13.17:9000",
            &roster,
            signed_health_by_other()
        ));

        // fp not in roster at all -> rejected, even with a correctly-signed health.
        let empty = RosterCache::from_entries(vec![]);
        assert!(!accept_peer_with(
            &fp,
            "http://10.13.13.17:9000",
            &empty,
            signed_health_ok(&key)
        ));
    }

    #[test]
    fn accept_peer_rejects_unreachable_or_unsigned_health() {
        let key = NodeKey::generate();
        let fp = key.fingerprint();
        let roster = RosterCache::from_entries(vec![(key.public_b64(), false)]);

        // health_fetch returning None (unreachable / malformed) -> rejected.
        let never = |_url: &str| -> Option<(String, String, Vec<u8>)> { None };
        assert!(!accept_peer_with(
            &fp,
            "http://10.13.13.17:9000",
            &roster,
            never
        ));
    }
}

#[cfg(test)]
mod advert_tests {
    use super::super::node_identity::NodeKey;
    use super::super::roster_cache::RosterCache;
    use super::super::worker::BrokerResources;
    use super::{verify_advert_body, Advert};

    fn advert_json(fp: &str) -> Vec<u8> {
        let advert = Advert {
            fp: fp.to_string(),
            price_per_hour: 3.6,
            resources: BrokerResources::default(),
            epoch: 1,
        };
        serde_json::to_vec(&advert).unwrap()
    }

    fn sign_ok(key: &NodeKey, body: &[u8]) -> String {
        key.sign(body)
    }

    #[test]
    fn fetch_advert_rejects_wrong_signer_and_unrostered() {
        let key = NodeKey::generate();
        let fp = key.fingerprint();
        let roster = RosterCache::from_entries(vec![(key.public_b64(), false)]);

        // advert.fp == signer fp, signer rostered -> Some
        let body_ok = advert_json(&fp);
        assert!(verify_advert_body(
            &body_ok,
            &key.public_b64(),
            &sign_ok(&key, &body_ok),
            &roster
        )
        .is_some());

        // advert claims a DIFFERENT fp than the signer -> None (impersonation).
        // Use a genuinely distinct key so this is a real crypto mismatch, not
        // just a string mismatch.
        let other = NodeKey::generate();
        let other_fp = other.fingerprint();
        let body_bad = advert_json(&other_fp);
        assert!(verify_advert_body(
            &body_bad,
            &key.public_b64(),
            &sign_ok(&key, &body_bad),
            &roster
        )
        .is_none());

        // signer not in roster -> None
        let empty = RosterCache::from_entries(vec![]);
        assert!(verify_advert_body(
            &body_ok,
            &key.public_b64(),
            &sign_ok(&key, &body_ok),
            &empty
        )
        .is_none());
    }

    /// Task 4: the discovery tick must refresh + cache each known peer's
    /// advert. Spins a minimal `/peer/advert` responder (same seam style as
    /// `fetch_gossip_signs_request_so_check_node_sig_gate_accepts_it`) and
    /// drives it through `refresh_adverts` end-to-end.
    #[test]
    fn discovery_round_caches_peer_adverts() {
        use crate::broker::node_identity::{verify_request, ReplayGuard};
        use crate::broker::worker::BrokerResources;
        use std::sync::Arc;

        let responder_key = NodeKey::generate();
        let responder_pub = responder_key.public_b64();
        let responder_fp = responder_key.fingerprint();

        let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
        let port = server.server_addr().to_ip().unwrap().port();
        std::thread::spawn(move || {
            let guard = ReplayGuard::new();
            for req in server.incoming_requests() {
                let headers: Vec<(String, String)> = req
                    .headers()
                    .iter()
                    .map(|h| (h.field.to_string(), h.value.to_string()))
                    .collect();
                let get = |name: &str| -> Option<String> {
                    headers
                        .iter()
                        .find(|(k, _)| k.eq_ignore_ascii_case(name))
                        .map(|(_, v)| v.clone())
                };
                let is_rostered = |_id: &str| true;
                let now = crate::broker::node_identity::now_secs();
                let verified =
                    verify_request(&is_rostered, &guard, "GET", "/peer/advert", b"", &get, now);

                let resp = match verified {
                    Ok(_) => {
                        let advert = super::Advert {
                            fp: responder_fp.clone(),
                            price_per_hour: 7.5,
                            resources: BrokerResources::default(),
                            epoch: 1,
                        };
                        let body = serde_json::to_vec(&advert).unwrap();
                        let sig = responder_key.sign(&body);
                        tiny_http::Response::from_data(body)
                            .with_status_code(200)
                            .with_header(
                                tiny_http::Header::from_bytes(
                                    "X-Node-Id",
                                    responder_key.public_b64(),
                                )
                                .unwrap(),
                            )
                            .with_header(tiny_http::Header::from_bytes("X-Node-Sig", sig).unwrap())
                    }
                    Err(_) => {
                        tiny_http::Response::from_string("unauthorized").with_status_code(401)
                    }
                };
                let _ = req.respond(resp);
            }
        });

        let peer_url = format!("http://127.0.0.1:{}", port);
        let roster = RosterCache::from_entries(vec![(responder_pub, false)]);
        let state = Arc::new(crate::broker::BrokerState::new());
        state.peer_manager.register_peer(peer_url.clone());

        super::refresh_adverts(&state, &roster);

        let a = state
            .peer_manager
            .peer_advert(&peer_url)
            .expect("advert should be cached after refresh");
        assert_eq!(a.price_per_hour, 7.5);
    }
}

/// Finding 1: the peer identity cache must be seeded eagerly at admission, not
/// only by the ~60s `health_check_all` tick, or every remote execution in the
/// window after a restart settles for free.
#[cfg(test)]
mod eager_fingerprint_seeding_tests {
    use super::super::node_identity::NodeKey;
    use super::super::roster_cache::RosterCache;
    use super::admit_verified_peer_with;
    use crate::broker::BrokerState;
    use std::sync::Arc;

    #[test]
    fn discovery_admission_seeds_fingerprint_without_any_health_probe() {
        let key = NodeKey::generate();
        let fp = key.fingerprint();
        let roster = RosterCache::from_entries(vec![(key.public_b64(), false)]);
        let state = Arc::new(BrokerState::new());
        let url = "http://10.13.13.17:9000";

        // Stubbed signed /peer/health, as `run_discovery_round` supplies it.
        // `PeerClient::check_health` is NEVER called in this test — no network.
        let body = b"{\"status\":\"healthy\"}".to_vec();
        let sig = key.sign(&body);
        let pubkey = key.public_b64();
        let accepted = admit_verified_peer_with(&state, &fp, url, &roster, move |_| {
            Some((pubkey.clone(), sig.clone(), body.clone()))
        });
        assert!(accepted);

        // The verified fingerprint is available for settlement IMMEDIATELY, and
        // is stored in the CANONICAL addressable form — `/peers` and `/brokers`'
        // `BrokerEntry.id` read `peer_fingerprint()` verbatim, so a bare hex
        // seed here would emit a non-addressable id until the ~60s health tick
        // rewrote it (finding 3).
        assert_eq!(
            state.peer_manager.peer_fingerprint(url).as_deref(),
            Some(format!("zc://node-{fp}").as_str()),
            "seeded identity must be canonical `zc://node-<fp>`, not bare hex"
        );
        assert_eq!(
            state.peer_manager.get_url_for_fingerprint(&fp).as_deref(),
            Some(url),
            "earn path must resolve this peer with no health probe having run"
        );
        // Settlement behaviour is unchanged by canonicalizing the STORED value:
        // `get_url_for_fingerprint` applies `strip_node_arg` to the stored side,
        // so a canonical `zc://node-<fp>` and a bare `<fp>` in the cache resolve
        // identically. The money path supplies bare hex (`executor_fp` returns
        // `Worker::node_fp`, stamped bare by `peer_node_fp` /
        // `NodeKey::fingerprint`) — asserted just above.
        //
        // The QUERY side is now normalized too, so the canonical form resolves
        // as well: the function is TOTAL over both identity forms. This
        // assertion previously pinned the OPPOSITE (bare-only, `None` for the
        // canonical form) as a deliberate record of a pre-existing trap; that
        // trap is now fixed, because a caller passing the canonical form — the
        // form `TaskResult::executor_node_uri` carries — was silently routed
        // into the pay-nobody settlement arm.
        assert_eq!(
            state
                .peer_manager
                .get_url_for_fingerprint(&format!("zc://node-{fp}"))
                .as_deref(),
            Some(url),
            "query side is normalized too: the canonical zc://node-<fp> form must \
             resolve, so a consumer of executor_node_uri cannot silently fall \
             into the free-work arm"
        );
    }

    #[test]
    fn rejected_peer_is_neither_registered_nor_seeded() {
        // Verification must still gate admission — seeding is not a bypass.
        let key = NodeKey::generate();
        let fp = key.fingerprint();
        let empty_roster = RosterCache::from_entries(vec![]);
        let state = Arc::new(BrokerState::new());
        let url = "http://10.13.13.18:9000";

        let body = b"{\"status\":\"healthy\"}".to_vec();
        let sig = key.sign(&body);
        let pubkey = key.public_b64();
        let accepted = admit_verified_peer_with(&state, &fp, url, &empty_roster, move |_| {
            Some((pubkey.clone(), sig.clone(), body.clone()))
        });
        assert!(!accepted, "unrostered fp must be rejected");
        assert_eq!(state.peer_manager.peer_fingerprint(url), None);
        assert_eq!(state.peer_manager.get_url_for_fingerprint(&fp), None);
    }
}

#[cfg(test)]
mod provider_field_sync_tests {
    use super::provider_fields_from_json;
    use crate::broker::worker::ProviderType;

    #[test]
    fn parses_provider_fields_off_peer_entry() {
        let w = serde_json::json!({
            "name": "zc-serve-abc",
            "provider_type": "general",
            "served_models": ["*"],
            "price_per_mtok": 2.0,
        });
        let (pt, models, price) = provider_fields_from_json(&w);
        assert_eq!(pt, ProviderType::General);
        assert_eq!(models, vec!["*".to_string()]);
        assert_eq!(price, 2.0);
    }

    #[test]
    fn absent_fields_fall_back_to_execute_worker_defaults() {
        // An older peer's /peer/workers entry has none of the fields; the
        // worker must sync exactly as before, just not as a model provider.
        let w = serde_json::json!({ "name": "worker-lxd" });
        let (pt, models, price) = provider_fields_from_json(&w);
        assert_eq!(pt, ProviderType::Specialized);
        assert!(models.is_empty());
        assert_eq!(price, 0.0);
    }

    #[test]
    fn malformed_fields_do_not_poison_the_sync() {
        let w = serde_json::json!({
            "provider_type": 7,
            "served_models": "not-a-list",
            "price_per_mtok": "free",
        });
        let (pt, models, price) = provider_fields_from_json(&w);
        assert_eq!(pt, ProviderType::Specialized);
        assert!(models.is_empty());
        assert_eq!(price, 0.0);
    }
}

#[cfg(test)]
mod peer_manager_worker_sync_tests {
    use super::resolve_dynamic_peer_broker_addrs;
    use std::collections::HashSet;

    /// THE BUG: a broker discovered dynamically (mesh-subnet scan, localhost
    /// scan, or gossip admission -- see mod.rs ~L320 / run_discovery_round)
    /// lands in `PeerManager` (hence the "[P2P] Registered peer broker" log
    /// line) but never in `DiscoveryConfig.peers`, which is sourced solely
    /// from `ZAKURO_PEERS`. `scan_peers` used to walk only the latter, so
    /// `/peer/workers` was never fetched from a mesh-discovered broker and
    /// `zc ls` showed 0 workers even though the peer broker's own
    /// `/peer/workers` returns them. This test captures the fix's pure
    /// merge logic: PeerManager-known broker URLs must show up in the
    /// address list to scan, with no ZAKURO_PEERS configured at all.
    #[test]
    fn dynamic_peer_manager_brokers_are_included_with_no_explicit_peers() {
        let dynamic_urls = vec![
            "http://10.13.13.11:9000".to_string(),
            "http://10.13.13.12:9000".to_string(),
        ];
        let explicit_hosts: HashSet<(String, u16)> = HashSet::new();

        let mut addrs = resolve_dynamic_peer_broker_addrs(&dynamic_urls, &explicit_hosts);
        addrs.sort();

        assert_eq!(
            addrs,
            vec![
                ("10.13.13.11".to_string(), 9000),
                ("10.13.13.12".to_string(), 9000),
            ]
        );
    }

    /// A broker present in BOTH ZAKURO_PEERS and PeerManager's dynamic list
    /// (e.g. gossip re-admits a broker the operator also named explicitly)
    /// must be scanned once per tick, not twice.
    #[test]
    fn already_explicit_dynamic_peer_is_not_duplicated() {
        let dynamic_urls = vec!["http://10.13.13.11:9000".to_string()];
        let mut explicit_hosts = HashSet::new();
        explicit_hosts.insert(("10.13.13.11".to_string(), 9000));

        let addrs = resolve_dynamic_peer_broker_addrs(&dynamic_urls, &explicit_hosts);
        assert!(addrs.is_empty());
    }

    /// Duplicate entries within the dynamic list itself (should not happen
    /// given PeerManager's DashMap keying, but never trust an invariant from
    /// another module without also being safe if it slips) must collapse to
    /// one scan, not N.
    #[test]
    fn duplicate_dynamic_urls_collapse_to_one_entry() {
        let dynamic_urls = vec![
            "http://10.13.13.11:9000".to_string(),
            "http://10.13.13.11:9000".to_string(),
        ];
        let explicit_hosts = HashSet::new();

        let addrs = resolve_dynamic_peer_broker_addrs(&dynamic_urls, &explicit_hosts);
        assert_eq!(addrs, vec![("10.13.13.11".to_string(), 9000)]);
    }

    /// Malformed URLs (missing port, empty host) must be skipped rather than
    /// panicking or poisoning the scan with a bogus address -- PeerManager is
    /// not expected to ever produce these, but this function must not trust
    /// that from the outside.
    #[test]
    fn malformed_dynamic_urls_are_skipped() {
        let dynamic_urls = vec![
            "http://".to_string(),
            "not-a-url".to_string(),
            "http://10.13.13.13:notaport".to_string(),
        ];
        let explicit_hosts = HashSet::new();

        let addrs = resolve_dynamic_peer_broker_addrs(&dynamic_urls, &explicit_hosts);
        assert!(addrs.is_empty());
    }
}