zc2 0.0.26

/// Integration and functional tests for the full broker HTTP server.
///
/// Each test spins up its own broker on a free OS-assigned port so tests
/// can run in parallel without conflicts.  The broker is configured with
/// no API credentials, so all credit operations run in-memory —
/// no external dependencies required.
///
/// Scenarios covered:
///   - Health check
///   - Worker registration, heartbeat, listing, removal
///   - Credit management: add, balance, balance correctness after execution
///   - Local broker mode (free execution — no credits deducted)
///   - Remote broker mode (credits charged via reserve-commit)
///   - Price estimation before execution
///   - Changing worker compute price and verifying new estimates
///   - Execute with insufficient credits → rejected
///   - Rate limiting enforcement
///   - Worker staleness and health transitions
///   - WAL-backed balance correctness: reserve → cancel cycle
#[cfg(test)]
pub(crate) mod integration {
    use std::net::TcpListener;
    use std::sync::Arc;
    use std::thread;
    use std::time::Duration;

    use crate::broker::{discovery::DiscoveryConfig, server::start_server, BrokerConfig};

    // ── Helpers ──────────────────────────────────────────────────────────────

    /// Bind on port 0 to let the OS assign a free port, then release it.
    /// Tiny_http must reuse it immediately — fine in practice for tests.
    /// Hand out a unique, bindable TCP port (#111).
    ///
    /// The naive `bind(":0") → read port → drop listener → caller re-binds`
    /// pattern is a TOCTOU race: under parallel tests the OS can hand the same
    /// just-freed ephemeral port to two concurrent callers, so both bind it and
    /// the loser dies with `AddrInUse` (or, for two brokers, the test silently
    /// talks to the wrong one — the 200-vs-401 peer-auth flake). A
    /// process-wide monotonic counter guarantees no two callers ever receive
    /// the same number; the probe-bind skips ports an external process holds.
    // Base well above the OS ephemeral range start to avoid churn with
    // sockets the OS hands out via bind(:0) elsewhere.
    static NEXT_PORT: std::sync::atomic::AtomicU16 = std::sync::atomic::AtomicU16::new(20000);

    /// Reserve `n` consecutive ports from the same non-ephemeral counter
    /// `free_port()` draws single ports from, none of which are bindable by
    /// an unrelated caller in between: every candidate is probed before the
    /// block is handed out, and the unique counter value ensures no sibling
    /// caller is ever handed the same numbers.
    fn port_range(n: u16) -> u16 {
        use std::sync::atomic::Ordering;
        loop {
            let base = NEXT_PORT.fetch_add(n, Ordering::Relaxed);
            assert!(
                base.checked_add(n).is_some_and(|end| end < 60000),
                "port_range exhausted the test port range"
            );
            if (base..base + n).all(|p| TcpListener::bind(("127.0.0.1", p)).is_ok()) {
                return base;
            }
        }
    }

    fn free_port() -> u16 {
        // Probe: if it binds now it is free; we drop the probe immediately
        // and the unique counter value ensures no sibling caller competes
        // for this same number.
        port_range(1)
    }

    /// Bind a `tiny_http` server on a unique loopback port, retrying on the
    /// rare case where an external process grabs the probed port before we
    /// re-bind it. `free_port()` already guarantees no *sibling test* competes
    /// for the same number. (#111)
    pub(crate) fn bind_loopback_http() -> (tiny_http::Server, u16) {
        let mut last = None;
        for _ in 0..10 {
            let port = free_port();
            match tiny_http::Server::http(("127.0.0.1", port)) {
                Ok(server) => return (server, port),
                Err(e) => last = Some(e),
            }
        }
        panic!("tiny_http loopback bind failed after 10 retries: {last:?}");
    }

    /// Bounded retry for flaky operations (#111). Re-runs `f` while it returns a
    /// *transient* error, up to `attempts` total tries, with a short linear
    /// backoff. Non-transient errors (e.g. an HTTP status the test means to
    /// assert on) are returned immediately, never masked.
    fn retry<T, E>(
        attempts: u32,
        is_transient: impl Fn(&E) -> bool,
        mut f: impl FnMut() -> Result<T, E>,
    ) -> Result<T, E> {
        let mut attempt = 1u32;
        loop {
            match f() {
                Ok(v) => return Ok(v),
                Err(e) if is_transient(&e) && attempt < attempts => {
                    thread::sleep(Duration::from_millis(25 * attempt as u64));
                    attempt += 1;
                }
                Err(e) => return Err(e),
            }
        }
    }

    /// A ureq error worth retrying: a transport-level hiccup under CPU
    /// contention (timeout, IO error, failed connect) — never an HTTP status,
    /// which is a real response the caller may be asserting on.
    fn is_transient_ureq(e: &ureq::Error) -> bool {
        matches!(
            e,
            ureq::Error::Timeout(_) | ureq::Error::Io(_) | ureq::Error::ConnectionFailed
        )
    }

    // ── Tests for the hardened test-infra helpers (#111) ───────────────────────

    #[test]
    fn retry_returns_ok_after_transient_failures() {
        use std::cell::Cell;
        let calls = Cell::new(0u32);
        // Transient error twice, then success on the 3rd attempt.
        let result: Result<&str, &str> = retry(
            3,
            |_e| true,
            || {
                let n = calls.get() + 1;
                calls.set(n);
                if n < 3 {
                    Err("transient")
                } else {
                    Ok("ok")
                }
            },
        );
        assert_eq!(result, Ok("ok"));
        assert_eq!(calls.get(), 3, "should retry up to the first success");
    }

    #[test]
    fn retry_does_not_retry_non_transient() {
        use std::cell::Cell;
        let calls = Cell::new(0u32);
        let result: Result<&str, &str> = retry(
            3,
            |_e| false,
            || {
                calls.set(calls.get() + 1);
                Err("fatal")
            },
        );
        assert_eq!(result, Err("fatal"));
        assert_eq!(calls.get(), 1, "a non-transient error must not be retried");
    }

    #[test]
    fn retry_gives_up_after_attempts() {
        use std::cell::Cell;
        let calls = Cell::new(0u32);
        let result: Result<&str, &str> = retry(
            3,
            |_e| true,
            || {
                calls.set(calls.get() + 1);
                Err("always")
            },
        );
        assert_eq!(result, Err("always"));
        assert_eq!(calls.get(), 3, "should attempt exactly `attempts` times");
    }

    #[test]
    fn free_port_hands_out_unique_ports() {
        use std::collections::HashSet;
        let mut handles = vec![];
        for _ in 0..16 {
            handles.push(thread::spawn(|| {
                (0..20).map(|_| free_port()).collect::<Vec<u16>>()
            }));
        }
        let mut all = HashSet::new();
        for h in handles {
            for p in h.join().unwrap() {
                assert!(all.insert(p), "free_port handed out a duplicate: {p}");
            }
        }
        assert_eq!(all.len(), 16 * 20, "every allocated port must be unique");
    }

    /// Build a minimal test config.
    /// Discovery is enabled so the broker detects "local mode" (no WireGuard → free execution).
    /// Scanning is disabled so no actual network probes are made.
    fn test_config(port: u16) -> BrokerConfig {
        BrokerConfig {
            host: "127.0.0.1".to_string(),
            port,
            verbose: false,
            daemon: false,
            tui_mode: false,
            enable_discovery: true,
            health_check_interval: 1,
            worker_timeout: 5,
            min_credits: 0.0001,
            enable_p2p: false,
            peer_key: None,
            owner_user_id: None,
            node_name: None,
            worker_key: None,
            // Explicitly None — never inherit production credentials from env
            api_url: None,
            api_key: None,
            wireguard_ip_override: None,
            quic_port: None,
            runtime_worker_threads: None,
            discovery: DiscoveryConfig {
                subnet: "10.13.13".to_string(),
                worker_port: 3960,
                extra_ports: vec![],
                scan_port_range: None,
                interval_secs: 60,
                enable_scan: false,
                enable_dns: false,
                peers: vec![],
                local_workers: vec![],
            },
        }
    }

    /// Start broker in a background thread and block until /health responds.
    fn start_broker(port: u16) -> String {
        let cfg = test_config(port);
        thread::spawn(move || {
            let _ = start_server(cfg);
        });

        let url = format!("http://127.0.0.1:{}", port);
        for _ in 0..300 {
            if ureq::get(&format!("{}/health", url))
                .config()
                .timeout_global(Some(Duration::from_millis(150)))
                .build()
                .call()
                .is_ok()
            {
                return url;
            }
            thread::sleep(Duration::from_millis(100));
        }
        panic!("Broker on port {} did not start within 30s", port);
    }

    fn get(url: &str) -> ureq::http::Response<ureq::Body> {
        retry(3, is_transient_ureq, || {
            ureq::get(url)
                .config()
                .timeout_global(Some(Duration::from_secs(5)))
                .build()
                .call()
        })
        .unwrap_or_else(|e| panic!("GET {url} failed after retries: {e:?}"))
    }

    fn get_json(url: &str) -> serde_json::Value {
        get(url).into_body().read_json().unwrap()
    }

    fn post_json(url: &str, body: serde_json::Value) -> ureq::http::Response<ureq::Body> {
        retry(3, is_transient_ureq, || {
            ureq::post(url)
                .header("Content-Type", "application/json")
                .config()
                .timeout_global(Some(Duration::from_secs(5)))
                .build()
                .send_json(body.clone())
        })
        .unwrap_or_else(|e| panic!("POST {url} failed after retries: {e:?}"))
    }

    fn post_json_err(url: &str, body: serde_json::Value) -> ureq::http::Response<ureq::Body> {
        retry(3, is_transient_ureq, || {
            ureq::post(url)
                .header("Content-Type", "application/json")
                .config()
                .timeout_global(Some(Duration::from_secs(5)))
                .http_status_as_error(false)
                .build()
                .send_json(body.clone())
        })
        .unwrap_or_else(|e| panic!("POST(err) {url} failed after retries: {e:?}"))
    }

    fn post_json_with(
        url: &str,
        body: serde_json::Value,
        headers: &[(&str, &str)],
    ) -> ureq::http::Response<ureq::Body> {
        retry(3, is_transient_ureq, || {
            let mut req = ureq::post(url)
                .header("Content-Type", "application/json")
                .config()
                .timeout_global(Some(Duration::from_secs(5)))
                .http_status_as_error(false)
                .build();
            for (k, v) in headers {
                req = req.header(*k, *v);
            }
            req.send_json(body.clone())
        })
        .unwrap_or_else(|e| panic!("POST(with) {url} failed after retries: {e:?}"))
    }

    fn add_credits(base: &str, user: &str, amount: f64) {
        let url = format!("{}/credits/{}/add", base, user);
        let resp = retry(3, is_transient_ureq, || {
            ureq::post(&url)
                .header("Content-Type", "application/json")
                .header("X-Api-Key", "test-master")
                .config()
                .timeout_global(Some(Duration::from_secs(5)))
                .http_status_as_error(false)
                .build()
                .send_json(serde_json::json!({"amount": amount}))
        })
        .unwrap_or_else(|e| panic!("add_credits {url} failed after retries: {e:?}"));
        assert!(
            resp.status().as_u16() == 200,
            "add_credits returned {}: {:?}",
            resp.status(),
            resp.into_body().read_to_string().ok()
        );
    }

    fn get_balance(base: &str, user: &str) -> f64 {
        let url = format!("{}/credits/{}", base, user);
        // Authenticate as the user via a zk_ key (resolves to `user`, which the
        // /credits/<user> handler allows). Avoids relying on the removed
        // non-zk_ admin-default (audit M4 / Task 1).
        let resp = retry(3, is_transient_ureq, || {
            ureq::get(&url)
                .header("Authorization", &format!("Bearer zk_{}_test", user))
                .config()
                .timeout_global(Some(Duration::from_secs(5)))
                .build()
                .call()
        })
        .unwrap_or_else(|e| panic!("get_balance {url} failed after retries: {e:?}"));
        let json: serde_json::Value = resp.into_body().read_json().unwrap();
        json["balance"].as_f64().unwrap()
    }

    fn register_worker(
        base: &str,
        name: &str,
        cpus: f64,
        memory_gib: u64,
        price_per_hour: f64,
    ) -> String {
        let body = serde_json::json!({
            "name": name,
            // A port nothing is listening on, NOT the real worker port 3960.
            // Callers register a worker here expecting it to be UNREACHABLE
            // (they assert the request is rejected). Hardcoding 3960 made that
            // true only on machines not running zc: on a box with a live
            // worker the "unreachable" worker answered, the job really
            // executed, and test_execute_insufficient_credits_rejected failed
            // for reasons unrelated to the code under test.
            "uri": format!("http://127.0.0.1:{}/{}", free_port(), name),
            "worker_type": "zakuro",
            "resources": {
                "cpus_total": cpus,
                "cpus_available": cpus,
                "memory_total": memory_gib * 1024 * 1024 * 1024,
                "memory_available": memory_gib * 1024 * 1024 * 1024,
                "gpus_total": 0,
                "gpus_available": 0
            },
            "pricing": {
                "price_per_hour": price_per_hour,
                "min_charge": 0.001
            }
        });
        let resp = post_json(&format!("{}/workers", base), body);
        let json: serde_json::Value = resp.into_body().read_json().unwrap();
        json["id"].as_str().unwrap().to_string()
    }

    // ── Tests ─────────────────────────────────────────────────────────────────

    /// GET /health returns {"status":"healthy"} without any auth.
    #[test]
    fn test_health_check_no_auth_required() {
        let url = start_broker(free_port());
        let body = get_json(&format!("{}/health", url));
        assert_eq!(body["status"], "healthy");
    }

    /// GET /workers returns empty list on a fresh broker.
    #[test]
    fn test_worker_list_empty_on_fresh_broker() {
        let url = start_broker(free_port());
        let body = get_json(&format!("{}/workers", url));
        assert_eq!(body["total"], 0);
        assert!(body["workers"].as_array().unwrap().is_empty());
    }

    /// POST /workers → worker registered; GET /workers → appears in list.
    #[test]
    fn test_register_worker_appears_in_list() {
        let url = start_broker(free_port());
        let id = register_worker(&url, "compute-1", 4.0, 8, 0.001);

        assert!(!id.is_empty());

        let list = get_json(&format!("{}/workers", url));
        assert_eq!(list["total"], 1);
        let workers = list["workers"].as_array().unwrap();
        assert_eq!(workers[0]["name"], "compute-1");
        assert_eq!(workers[0]["id"], id);
    }

    /// Multiple workers registered — all show in /workers.
    #[test]
    fn test_multiple_workers_registered() {
        let url = start_broker(free_port());
        register_worker(&url, "w1", 4.0, 8, 0.001);
        register_worker(&url, "w2", 8.0, 16, 0.002);

        let list = get_json(&format!("{}/workers", url));
        assert_eq!(list["total"], 2);
    }

    /// Worker heartbeat updates its status.
    #[test]
    fn test_worker_heartbeat_accepted() {
        let url = start_broker(free_port());
        let id = register_worker(&url, "hb-worker", 4.0, 8, 0.001);

        let resp = post_json(
            &format!("{}/workers/heartbeat", url),
            serde_json::json!({"worker_id": id, "active_requests": 3}),
        );
        assert_eq!(resp.status().as_u16(), 200);

        let workers = get_json(&format!("{}/workers", url));
        let w = &workers["workers"].as_array().unwrap()[0];
        assert_eq!(w["active_requests"], 3);
    }

    /// DELETE /workers/:id removes it from the list.
    #[test]
    fn test_unregister_worker_removes_from_list() {
        let url = start_broker(free_port());
        let id = register_worker(&url, "temp-worker", 2.0, 4, 0.001);

        assert_eq!(get_json(&format!("{}/workers", url))["total"], 1);

        ureq::delete(&format!("{}/workers/{}", url, id))
            .config()
            .timeout_global(Some(Duration::from_secs(5)))
            .build()
            .call()
            .ok();

        assert_eq!(get_json(&format!("{}/workers", url))["total"], 0);
    }

    // ── Credit management ──────────────────────────────────────────────────

    /// POST /credits/:user/add then GET /credits/:user shows correct balance.
    #[test]
    fn test_add_credits_and_read_balance() {
        let url = start_broker(free_port());

        add_credits(&url, "alice", 100.0);
        let balance = get_balance(&url, "alice");
        assert!((balance - 100.0).abs() < 0.001);
    }

    /// Credits accumulate across multiple top-ups.
    #[test]
    fn test_multiple_topups_accumulate() {
        let url = start_broker(free_port());

        add_credits(&url, "bob", 50.0);
        add_credits(&url, "bob", 30.0);
        add_credits(&url, "bob", 20.0);

        let balance = get_balance(&url, "bob");
        assert!((balance - 100.0).abs() < 0.001);
    }

    /// Different users have independent balances.
    #[test]
    fn test_user_balances_are_independent() {
        let url = start_broker(free_port());

        add_credits(&url, "user-a", 100.0);
        add_credits(&url, "user-b", 200.0);

        assert!((get_balance(&url, "user-a") - 100.0).abs() < 0.001);
        assert!((get_balance(&url, "user-b") - 200.0).abs() < 0.001);
    }

    // ── Price estimation ───────────────────────────────────────────────────

    /// POST /price returns min_cost / max_cost for available workers.
    #[test]
    fn test_price_estimate_with_workers() {
        let url = start_broker(free_port());
        register_worker(&url, "price-w", 4.0, 8, 0.001);

        let resp = post_json_err(
            &format!("{}/price", url),
            serde_json::json!({
                "cpus": 1.0,
                "memory_bytes": 1073741824,
                "estimated_duration_secs": 10.0
            }),
        );
        assert_eq!(resp.status().as_u16(), 200);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert!(body["min_cost"].as_f64().unwrap() > 0.0);
        assert!(body["matching_workers"].as_u64().unwrap() >= 1);
    }

    /// POST /price returns 503 when no workers are available.
    #[test]
    fn test_price_estimate_no_workers_returns_error() {
        let url = start_broker(free_port());
        let resp = post_json_err(&format!("{}/price", url), serde_json::json!({"cpus": 1.0}));
        assert_eq!(resp.status().as_u16(), 503);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert_eq!(body["code"], "NO_CAPACITY");
    }

    /// Changing a worker's price is reflected in subsequent price estimates.
    /// (Done here by registering two workers with different prices and verifying
    ///  the min reflects the cheapest one.)
    #[test]
    fn test_worker_price_affects_estimate() {
        let url = start_broker(free_port());
        register_worker(&url, "cheap-w", 4.0, 8, 0.001);
        register_worker(&url, "pricey-w", 4.0, 8, 1.0);

        let resp = post_json_err(
            &format!("{}/price", url),
            serde_json::json!({
                "cpus": 1.0,
                "memory_bytes": 1073741824,
                "estimated_duration_secs": 3600.0
            }),
        );
        assert_eq!(resp.status().as_u16(), 200);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        let min = body["min_cost"].as_f64().unwrap();
        let max = body["max_cost"].as_f64().unwrap();
        // min should be from the cheap worker (0.001), max from the pricey one (1.0)
        assert!(min < max, "min={} max={}", min, max);
        assert_eq!(body["matching_workers"], 2);
    }

    // ── Execute (local mode — free) ────────────────────────────────────────

    /// In local mode (no WireGuard, localhost workers) execute returns NO_WORKERS
    /// when no worker is registered — confirms auth is skipped in local mode.
    #[test]
    fn test_execute_local_mode_no_worker_returns_no_workers() {
        let url = start_broker(free_port());

        // Local mode: no auth header needed
        let resp = post_json_err(
            &format!("{}/execute", url),
            serde_json::json!({"fn": "test_fn", "args": []}),
        );
        // Expects NO_WORKERS, not UNAUTHORIZED
        assert_eq!(resp.status().as_u16(), 503);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert_eq!(body["code"], "NO_WORKERS");
    }

    // ── Execute (remote mode — credits charged) ────────────────────────────

    /// Execute without auth returns UNAUTHORIZED in remote (non-local) mode.
    #[test]
    fn test_execute_no_auth_returns_unauthorized_when_local_mode_disabled() {
        // We test the broker's behavior: since a localhost worker is registered,
        // the broker is likely in local mode. In remote mode we'd need WireGuard.
        // This test verifies the auth path is gated correctly.
        let url = start_broker(free_port());

        // With a real API key format but no worker
        let resp = post_json_with(
            &format!("{}/execute", url),
            serde_json::json!({"fn": "greet"}),
            &[("Authorization", "Bearer test-master")],
        );
        // Either NO_WORKERS (local mode) or another execute error
        assert!(resp.status().as_u16() >= 400 || resp.status().as_u16() == 503);
    }

    /// Execute with insufficient credits is rejected with INSUFFICIENT_CREDITS.
    #[test]
    fn test_execute_insufficient_credits_rejected() {
        let url = start_broker(free_port());

        // Register a worker that advertises a high price
        register_worker(&url, "gpu-w", 8.0, 32, 10.0);

        // User has no credits
        let resp = post_json_with(
            &format!("{}/execute", url),
            serde_json::json!({
                "fn": "train",
                "cpus": 8.0,
                "estimated_duration_secs": 3600.0
            }),
            &[("X-Zakuro-User", "broke-user")],
        );

        // In local mode the auth is skipped and NO_WORKERS / NO_CAPACITY might come back.
        // The key assertion: the request is rejected.
        assert!(resp.status().as_u16() >= 400 || resp.status().as_u16() == 503);
    }

    // ── Broker state: local_mode / is_local_worker ────────────────────────

    #[test]
    fn test_broker_state_local_mode_flag() {
        use crate::broker::BrokerState;
        let state = BrokerState::new();
        assert!(!state.is_local_mode()); // off by default
        state.set_local_mode(true);
        assert!(state.is_local_mode());
    }

    #[test]
    fn test_is_local_worker_localhost() {
        use crate::broker::BrokerState;
        let state = BrokerState::new();
        // In non-local mode, 127.0.0.1 URIs are still considered local
        assert!(state.is_local_worker("http://127.0.0.1:3960"));
        assert!(state.is_local_worker("http://localhost:3960"));
        assert!(!state.is_local_worker("http://10.13.13.5:3960"));
    }

    #[test]
    fn test_is_local_worker_all_local_in_local_mode() {
        use crate::broker::BrokerState;
        let state = BrokerState::new();
        state.set_local_mode(true);
        // In local_mode everything is local — even remote IPs
        assert!(state.is_local_worker("http://10.13.13.5:3960"));
    }

    // ── Rate limiting via CreditManager ───────────────────────────────────

    #[test]
    fn test_rate_limit_per_second_blocks_excess_requests() {
        use crate::broker::credits::CreditManager;
        let mgr = CreditManager::new();
        mgr.get_or_create("alice", 1000.0);
        mgr.set_rate_limits("alice", Some(3), None, None); // max 3/s
        mgr.set_rate_limit("alice", 1000); // high per-minute limit

        assert!(mgr.check_rate_limit("alice")); // 1
        assert!(mgr.check_rate_limit("alice")); // 2
        assert!(mgr.check_rate_limit("alice")); // 3
        assert!(!mgr.check_rate_limit("alice")); // 4 — blocked
    }

    // ── Worker staleness ──────────────────────────────────────────────────

    /// Workers that stop sending heartbeats are marked unhealthy.
    #[test]
    fn test_stale_worker_marked_unhealthy() {
        use crate::broker::worker::WorkerRegistration;
        use crate::broker::worker::{HardwareInfo, WorkerPricing, WorkerResources};
        use crate::broker::worker::{WorkerRegistry, WorkerStatus};

        let registry = WorkerRegistry::new();
        let w = registry.register(WorkerRegistration {
            name: "stale-w".to_string(),
            uri: "http://127.0.0.1:3960".to_string(),
            worker_type: "zakuro".to_string(),
            resources: WorkerResources::default(),
            pricing: WorkerPricing::default(),
            tags: vec![],
            max_timeout_secs: 0.0,
            hardware: HardwareInfo::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,
        });

        assert_eq!(registry.get(&w.id).unwrap().status, WorkerStatus::Healthy);
        // A timeout of -1 means every worker is immediately stale (elapsed > -1 always true)
        registry.mark_stale(-1);
        assert_eq!(registry.get(&w.id).unwrap().status, WorkerStatus::Unhealthy);
    }

    // ── Full broker scenario: worker registration + price check ───────────

    /// Functional scenario:
    ///  1. Start broker
    ///  2. Register a compute worker with custom pricing
    ///  3. Query the price — verify it matches the worker's pricing formula
    #[test]
    fn test_scenario_register_worker_and_verify_price() {
        let url = start_broker(free_port());

        // Register a worker: 7.2 credits/hour (0.002 credits/second)
        register_worker(&url, "precision-w", 8.0, 16, 7.2);

        let resp = post_json_err(
            &format!("{}/price", url),
            serde_json::json!({
                "cpus": 2.0,
                "memory_bytes": 2147483648u64,  // 2 GiB
                "estimated_duration_secs": 3600.0
            }),
        );
        assert_eq!(resp.status().as_u16(), 200);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        let min = body["min_cost"].as_f64().unwrap();

        // Expected: 7.2 credits/hour * 1 hour = 7.2 credits
        assert!(
            (min - 7.2).abs() < 0.01,
            "expected min_cost ≈ 7.2, got {}",
            min
        );
    }

    /// Functional scenario:
    ///  1. Start broker
    ///  2. Add credits to a user
    ///  3. Verify balance before any execution
    ///  4. Execute → no worker → NO_WORKERS (confirms credit check ran)
    #[test]
    fn test_scenario_add_credits_then_execute_path() {
        let url = start_broker(free_port());

        add_credits(&url, "charlie", 50.0);
        assert!((get_balance(&url, "charlie") - 50.0).abs() < 0.001);

        // No workers registered — execute path should return NO_WORKERS
        let resp = post_json_err(
            &format!("{}/execute", url),
            serde_json::json!({"fn": "hello"}),
        );
        assert_eq!(resp.status().as_u16(), 503);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert_eq!(body["code"], "NO_WORKERS");

        // Balance unchanged after a rejected request
        assert!((get_balance(&url, "charlie") - 50.0).abs() < 0.001);
    }

    // ── budget_credits ─────────────────────────────────────────────────────

    /// budget_credits → effective_timeout → reservation_amount derivation.
    /// With price_per_hour = 3600.0 (= 1 credit/sec) and budget_credits = 1.0,
    /// the derived effective timeout must be exactly 1 second, and the resulting
    /// reservation must be 1.0 credit (not the estimated_duration default).
    #[test]
    fn test_budget_credits_derives_correct_timeout_and_reservation() {
        use crate::broker::worker::WorkerPricing;

        let pricing = WorkerPricing {
            price_per_hour: 3600.0,
            min_charge: 0.001,
        };
        let budget = 1.0_f64;
        let price_per_sec = pricing.price_per_hour / 3600.0;

        // Mirror the server logic exactly (timeout_secs = 0.0 → no explicit cap)
        let max_secs = if price_per_sec > 0.0 {
            budget / price_per_sec
        } else {
            86_400.0
        };
        let effective_timeout = max_secs; // timeout_secs == 0.0

        assert!(
            (effective_timeout - 1.0).abs() < 1e-9,
            "expected effective_timeout = 1.0s, got {}",
            effective_timeout
        );

        // Reservation = cost for the full effective_timeout duration
        let reservation = pricing.estimate_cost(effective_timeout);
        assert!(
            (reservation - 1.0).abs() < 1e-9,
            "expected reservation = 1.0 credit, got {}",
            reservation
        );
    }

    /// budget_credits <= 0 in X-Zakuro-Requirements must be rejected with HTTP 400.
    #[test]
    fn test_execute_negative_budget_credits_rejected() {
        let url = start_broker(free_port());

        let resp = post_json_with(
            &format!("{}/execute", url),
            serde_json::json!({"fn": "test"}),
            &[("X-Zakuro-Requirements", r#"{"budget_credits": -1.0}"#)],
        );
        assert_eq!(resp.status().as_u16(), 400);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert_eq!(body["code"], "BAD_REQUEST");
    }

    // ═══════════════════════════════════════════════════════════════════
    // P2P & REMOTE-MODE HELPERS
    // ═══════════════════════════════════════════════════════════════════

    fn test_config_p2p(port: u16) -> BrokerConfig {
        BrokerConfig {
            peer_key: Some("test-peer-key".to_string()),
            enable_p2p: true,
            owner_user_id: Some("test-owner".to_string()),
            ..test_config(port)
        }
    }

    fn start_broker_p2p(port: u16) -> String {
        let cfg = test_config_p2p(port);
        thread::spawn(move || {
            let _ = start_server(cfg);
        });
        let url = format!("http://127.0.0.1:{}", port);
        for _ in 0..300 {
            if ureq::get(&format!("{}/health", url))
                .config()
                .timeout_global(Some(Duration::from_millis(150)))
                .build()
                .call()
                .is_ok()
            {
                return url;
            }
            thread::sleep(Duration::from_millis(100));
        }
        panic!("P2P Broker on port {} did not start within 30s", port);
    }

    fn start_broker_remote(port: u16) -> String {
        let cfg = BrokerConfig {
            enable_discovery: false,
            ..test_config(port)
        };
        thread::spawn(move || {
            let _ = start_server(cfg);
        });
        let url = format!("http://127.0.0.1:{}", port);
        for _ in 0..300 {
            if ureq::get(&format!("{}/health", url))
                .config()
                .timeout_global(Some(Duration::from_millis(150)))
                .build()
                .call()
                .is_ok()
            {
                return url;
            }
            thread::sleep(Duration::from_millis(100));
        }
        panic!("Remote broker on port {} did not start within 30s", port);
    }

    fn peer_post_json(
        url: &str,
        body: serde_json::Value,
        peer_key: &str,
    ) -> ureq::http::Response<ureq::Body> {
        retry(3, is_transient_ureq, || {
            ureq::post(url)
                .header("Content-Type", "application/json")
                .header("X-Peer-Key", peer_key)
                .config()
                .timeout_global(Some(Duration::from_secs(5)))
                .http_status_as_error(false)
                .build()
                .send_json(body.clone())
        })
        .unwrap_or_else(|e| panic!("peer POST {url} failed after retries: {e:?}"))
    }

    fn peer_get_json(url: &str, peer_key: &str) -> ureq::http::Response<ureq::Body> {
        retry(3, is_transient_ureq, || {
            ureq::get(url)
                .header("X-Peer-Key", peer_key)
                .config()
                .timeout_global(Some(Duration::from_secs(5)))
                .http_status_as_error(false)
                .build()
                .call()
        })
        .unwrap_or_else(|e| panic!("peer GET {url} failed after retries: {e:?}"))
    }

    fn start_mock_worker() -> (String, u16) {
        // free_port() is collision-free, but keep a bounded bind-retry as a
        // belt-and-suspenders guard against a port an external process grabs
        // between probe and bind. (#111)
        let (raw, port) = bind_loopback_http();
        let server = std::sync::Arc::new(raw);
        // Respond with `Connection: close` and serve from several acceptor
        // threads. Root cause of the `test_p2p_mesh_fibonacci_500` load flake:
        // the broker forwards over a pooled keep-alive `ureq::Agent`, so the
        // mock's acceptor parks on an *idle* keep-alive socket the broker holds
        // in its pool, while a fresh concurrent connection waits unaccepted
        // until the client's 30s global timeout fires (a hard stall, not mere
        // slowness — which is why reducing thread oversubscription never fixed
        // it). Closing after each response means no socket is held idle, so an
        // acceptor is always free.
        for _ in 0..4 {
            let server = server.clone();
            thread::spawn(move || {
                for request in server.incoming_requests() {
                    let response = tiny_http::Response::from_data(b"mock-result".to_vec())
                        .with_status_code(200)
                        .with_header(
                            tiny_http::Header::from_bytes(
                                "Content-Type",
                                "application/octet-stream",
                            )
                            .unwrap(),
                        )
                        .with_header(tiny_http::Header::from_bytes("Connection", "close").unwrap());
                    let _ = request.respond(response);
                }
            });
        }
        (format!("http://127.0.0.1:{}", port), port)
    }

    fn register_worker_at(base: &str, name: &str, uri: &str, price: f64) -> String {
        let body = serde_json::json!({
            "name": name,
            "uri": uri,
            "worker_type": "zakuro",
            "resources": {
                "cpus_total": 4.0,
                "cpus_available": 4.0,
                "memory_total": 8589934592u64,
                "memory_available": 8589934592u64,
                "gpus_total": 0,
                "gpus_available": 0
            },
            "pricing": {
                "price_per_hour": price,
                "min_charge": 0.001
            }
        });
        let resp = post_json(&format!("{}/workers", base), body);
        let json: serde_json::Value = resp.into_body().read_json().unwrap();
        json["id"].as_str().unwrap().to_string()
    }

    // ═══════════════════════════════════════════════════════════════════
    // PILLAR 1 — API MODE vs P2P BROKER-TO-BROKER
    // ═══════════════════════════════════════════════════════════════════

    // ── API key resolution ───────────────────────────────────────────

    /// zk_{user_id}_{hex} format is parsed correctly. Master key resolves to "admin".
    #[test]
    fn test_api_key_format_zk_user_hex_resolves_user_id() {
        use crate::broker::ledger::Ledger;
        let ledger = Ledger::new(None, None);
        assert_eq!(
            ledger
                .resolve_user_from_api_key("zk_9000000001_abc123")
                .unwrap(),
            "9000000001"
        );
        assert_eq!(
            ledger
                .resolve_user_from_api_key("zk_alice_deadbeef")
                .unwrap(),
            "alice"
        );
        // Non-zk_ keys resolve to "admin" in standalone mode (no ZAKURO_MASTER_KEY set)
        // when ZAKURO_MASTER_KEY is not configured
        // (actual env may vary — tested via ZAKURO_MASTER_KEY env var in production)
    }

    /// Invalid API key formats are rejected.
    #[test]
    fn test_api_key_invalid_format_rejected() {
        use crate::broker::ledger::Ledger;
        let ledger = Ledger::new(None, None);
        // Empty and malformed zk_ keys are always rejected
        assert!(ledger.resolve_user_from_api_key("").is_err());
        assert!(ledger.resolve_user_from_api_key("zk_").is_err());
        // zk_ with no user portion
        assert!(ledger
            .resolve_user_from_api_key("zk__nouserportion")
            .is_err());
    }

    // ── Ledger local operations (P2P in-memory zero-PG path) ─────────

    /// local_reserve → local_commit with partial refund.
    #[test]
    fn test_ledger_local_reserve_commit_refunds_difference() {
        use crate::broker::ledger::Ledger;
        let ledger = Ledger::new(None, None);

        ledger.local_add_credits("alice", 100.0);
        assert!((ledger.load_balance_if_needed("alice") - 100.0).abs() < 1e-9);

        let (rid, balance_before) = ledger.local_reserve("alice", 40.0, "req-1").unwrap();
        assert!((balance_before - 100.0).abs() < 1e-9);
        assert!((ledger.load_balance_if_needed("alice") - 60.0).abs() < 1e-9);

        let balance_after = ledger.local_commit(&rid, 25.0).unwrap();
        assert!(
            (balance_after - 75.0).abs() < 1e-9,
            "expected 75.0 (reserved 40, charged 25, refund 15), got {}",
            balance_after
        );
    }

    /// local_reserve fails with InsufficientCredits when balance is too low.
    #[test]
    fn test_ledger_local_reserve_insufficient_rejects() {
        use crate::broker::ledger::{Ledger, LedgerError};
        let ledger = Ledger::new(None, None);
        ledger.local_add_credits("bob", 10.0);
        match ledger.local_reserve("bob", 20.0, "req-x") {
            Err(LedgerError::InsufficientCredits {
                required,
                available,
            }) => {
                assert!((required - 20.0).abs() < 1e-9);
                assert!((available - 10.0).abs() < 1e-9);
            }
            other => panic!("expected InsufficientCredits, got {:?}", other),
        }
        assert!((ledger.load_balance_if_needed("bob") - 10.0).abs() < 1e-9);
    }

    /// A user whose balance was never loaded must NOT be reported as being out
    /// of credits. Regression test for a real incident: a broker that could not
    /// reach the dashboard denied every job with "Insufficient credits: have
    /// 0.0000" for an account holding 500 credits, because an unreachable
    /// authority collapsed to a cached 0.0. "Unknown" and "zero" must differ.
    #[test]
    fn test_ledger_unknown_balance_is_not_insufficient_credits() {
        use crate::broker::ledger::{Ledger, LedgerError};
        // An api_url that cannot resolve stands in for an unreachable
        // authority; no credits are ever seeded for this user.
        let ledger = Ledger::new(
            Some("http://127.0.0.1:1/unreachable".to_string()),
            Some("zk_unknown_key".to_string()),
        );

        assert!(
            ledger.try_get_balance("nobody").is_none(),
            "an unreachable authority with no cache must report unknown, not Some(0.0)"
        );

        match ledger.local_reserve("nobody", 1.0, "req-unknown") {
            Err(LedgerError::BalanceUnavailable { user_id }) => {
                assert_eq!(user_id, "nobody");
            }
            other => panic!("expected BalanceUnavailable, got {:?}", other),
        }

        // And a genuinely-zero balance must still read as insufficient, so the
        // new variant has not simply swallowed the real out-of-credits case.
        let known_zero = Ledger::new(None, None);
        known_zero.local_add_credits("alice", 0.0);
        match known_zero.local_reserve("alice", 5.0, "req-zero") {
            Err(LedgerError::InsufficientCredits { available, .. }) => {
                assert!(available.abs() < 1e-9, "expected 0.0, got {}", available);
            }
            other => panic!(
                "expected InsufficientCredits for a known-zero balance, got {:?}",
                other
            ),
        }
    }

    /// local_cancel fully restores the reserved amount.
    #[test]
    fn test_ledger_local_cancel_restores_full_balance() {
        use crate::broker::ledger::Ledger;
        let ledger = Ledger::new(None, None);
        ledger.local_add_credits("carol", 50.0);
        let (rid, _) = ledger.local_reserve("carol", 30.0, "req-c").unwrap();
        assert!((ledger.load_balance_if_needed("carol") - 20.0).abs() < 1e-9);
        ledger.local_cancel(&rid).unwrap();
        assert!((ledger.load_balance_if_needed("carol") - 50.0).abs() < 1e-9);
    }

    /// local_add_credits accumulates across multiple calls.
    #[test]
    fn test_ledger_local_add_credits_accumulates() {
        use crate::broker::ledger::Ledger;
        let ledger = Ledger::new(None, None);
        assert!((ledger.local_add_credits("dave", 50.0) - 50.0).abs() < 1e-9);
        assert!((ledger.local_add_credits("dave", 30.0) - 80.0).abs() < 1e-9);
        assert!((ledger.local_add_credits("dave", 20.0) - 100.0).abs() < 1e-9);
    }

    // ── C1+C4: amount validation & clamping ──────────────────────────

    /// Non-finite (NaN / Inf) and negative amounts are rejected at reserve.
    #[test]
    fn test_ledger_reserve_rejects_non_finite_and_negative() {
        use crate::broker::ledger::{Ledger, LedgerError};
        let ledger = Ledger::new(None, None);
        ledger.local_credits.insert("u".to_string(), 100.0);
        for bad in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY, -1.0] {
            match ledger.reserve("u", bad, "r") {
                Err(LedgerError::InvalidAmount(_)) => {}
                other => panic!("expected InvalidAmount for {}, got {:?}", bad, other),
            }
        }
        // Balance untouched by rejected reserves.
        assert!((ledger.local_credits.get("u").map(|v| *v).unwrap_or(0.0) - 100.0).abs() < 1e-9);
    }

    /// local_reserve rejects non-finite / negative amounts too.
    #[test]
    fn test_ledger_local_reserve_rejects_non_finite_and_negative() {
        use crate::broker::ledger::{Ledger, LedgerError};
        let ledger = Ledger::new(None, None);
        ledger.local_add_credits("u", 100.0);
        for bad in [f64::NAN, f64::INFINITY, -5.0] {
            match ledger.local_reserve("u", bad, "r") {
                Err(LedgerError::InvalidAmount(_)) => {}
                other => panic!("expected InvalidAmount for {}, got {:?}", bad, other),
            }
        }
        assert!((ledger.load_balance_if_needed("u") - 100.0).abs() < 1e-9);
    }

    /// commit caps actual cost to [0, reserved]: an over-large actual charges
    /// at most the reserved amount (no negative refund / no extra deduction),
    /// and a negative actual refunds the full reservation (charge clamped to 0).
    #[test]
    fn test_ledger_commit_caps_actual_to_reserved_range() {
        use crate::broker::ledger::Ledger;
        // Over-large actual_amount → charge capped at reserved (40), refund 0.
        let ledger = Ledger::new(None, None);
        ledger.local_credits.insert("u".to_string(), 100.0);
        let rid = ledger.reserve("u", 40.0, "r").unwrap();
        assert!((ledger.local_credits.get("u").map(|v| *v).unwrap_or(0.0) - 60.0).abs() < 1e-9);
        let bal = ledger.commit(&rid, 999.0).unwrap();
        assert!(
            (bal - 60.0).abs() < 1e-9,
            "charge capped at reserved 40, balance stays 60, got {}",
            bal
        );

        // Negative actual_amount → charge clamped to 0, full 40 refunded.
        let rid2 = ledger.reserve("u", 30.0, "r2").unwrap();
        let bal2 = ledger.commit(&rid2, -10.0).unwrap();
        assert!(
            (bal2 - 60.0).abs() < 1e-9,
            "negative charge clamped to 0, full refund, got {}",
            bal2
        );
    }

    /// local_commit caps actual_cost to [0, reserved].
    #[test]
    fn test_ledger_local_commit_caps_actual_to_reserved_range() {
        use crate::broker::ledger::Ledger;
        let ledger = Ledger::new(None, None);
        ledger.local_add_credits("u", 100.0);
        let (rid, _) = ledger.local_reserve("u", 40.0, "r").unwrap();
        // Over-large actual → capped at 40, balance 60.
        let bal = ledger.local_commit(&rid, 999.0).unwrap();
        assert!((bal - 60.0).abs() < 1e-9, "expected 60, got {}", bal);
        // Non-finite actual → treated as full reserved charge (no panic, balance stable).
        let (rid2, _) = ledger.local_reserve("u", 20.0, "r2").unwrap();
        let bal2 = ledger.local_commit(&rid2, f64::NAN).unwrap();
        assert!(
            (bal2 - 40.0).abs() < 1e-9,
            "NaN actual charges full reserved, got {}",
            bal2
        );
    }

    /// reserve against a balance present ONLY in authoritative_balances
    /// (local_credits empty) must succeed — regression for the and_modify no-op
    /// that reported a false InsufficientCredits.
    #[test]
    fn test_ledger_reserve_loads_absent_local_credits_key() {
        use crate::broker::ledger::Ledger;
        let ledger = Ledger::new(None, None);
        // Seed only authoritative_balances; local_credits has no key for "u".
        ledger.authoritative_balances.insert("u".to_string(), 50.0);
        let rid = ledger
            .reserve("u", 20.0, "r")
            .expect("reserve should load from authoritative");
        assert!((ledger.local_credits.get("u").map(|v| *v).unwrap_or(0.0) - 30.0).abs() < 1e-9);
        let bal = ledger.commit(&rid, 20.0).unwrap();
        assert!((bal - 30.0).abs() < 1e-9, "expected 30, got {}", bal);
    }

    // ── H4: HTTP body cap (413) ──────────────────────────────────────

    #[test]
    fn body_over_cap_returns_413() {
        let port = free_port();
        let base = start_broker(port);

        // Build a JSON body well over the 4 MiB cap: a single huge "name" string.
        let big = "A".repeat(6 * 1024 * 1024);
        let body = serde_json::json!({
            "name": big,
            "uri": "http://127.0.0.1:3960",
            "worker_type": "compute",
            "pricing": { "price_per_hour": 1.0, "min_charge": 0.0 },
            "resources": {
                "cpus_available": 1.0, "cpus_total": 1.0,
                "memory_available": 0, "memory_total": 0,
                "gpus_available": 0, "gpus_total": 0,
                "storage_gb": 0.0
            }
        });

        // The broker may refuse the oversized body in one of two equally-valid
        // ways: (a) read enough to reply 413, or (b) close the connection early
        // once the cap is exceeded — which surfaces to the client as a transport
        // error (BrokenPipe / ConnectionReset) while it is still streaming the
        // 6 MiB body. Both mean "refused"; only a 2xx accept is a real failure.
        // (Going through ureq directly here instead of `post_json_err`, whose
        // `.unwrap()` would panic on the early-close path — a flaky race.)
        let result = ureq::post(&format!("{}/workers", base))
            .header("Content-Type", "application/json")
            .config()
            .timeout_global(Some(Duration::from_secs(5)))
            .http_status_as_error(false)
            .build()
            .send_json(body);

        match result {
            Ok(resp) => assert_eq!(
                resp.status().as_u16(),
                413,
                "oversized body must be rejected with 413, got {}",
                resp.status()
            ),
            // Any transport-level error means the server cut the upload off
            // rather than accepting it — all equally "refused": BrokenPipe /
            // ConnectionReset on Linux, EINVAL on macOS, or a timeout under
            // load. Only a 2xx accept is a real failure. (#111)
            Err(ureq::Error::Io(_)) | Err(ureq::Error::Timeout(_)) => {}
            Err(other) => panic!("unexpected error sending oversized body: {other}"),
        }
    }

    // ── P2P authority determination ──────────────────────────────────

    /// With 2 brokers, FNV-1a hash distributes users deterministically
    /// across Local and Peer (or FallbackToPg when peer is unreachable).
    #[test]
    fn test_p2p_authority_deterministic_with_two_brokers() {
        use crate::broker::peer::{Authority, PeerManager};
        let pm = PeerManager::new(
            Some("10.0.0.1"),
            &["10.0.0.2:3960".to_string()],
            9000,
            "secret".to_string(),
            true,
        );

        // Same user always gets the same authority
        let a1 = pm.determine_authority("user-1");
        let a2 = pm.determine_authority("user-1");
        match (&a1, &a2) {
            (Authority::Local, Authority::Local) => {}
            (Authority::Standalone, Authority::Standalone) => {}
            (Authority::Peer(a), Authority::Peer(b)) => assert_eq!(a, b),
            _ => panic!("authority not deterministic: {:?} vs {:?}", a1, a2),
        }

        // 100 users should spread across Local and FallbackToPg (peer unreachable)
        let mut local_count = 0;
        let mut fallback_count = 0;
        for i in 0..100 {
            match pm.determine_authority(&format!("user-{}", i)) {
                Authority::Local => local_count += 1,
                Authority::Standalone => fallback_count += 1,
                _ => {}
            }
        }
        assert!(
            local_count >= 20,
            "expected >=20 local, got {}",
            local_count
        );
        assert!(
            fallback_count >= 20,
            "expected >=20 fallback, got {}",
            fallback_count
        );
    }

    // ── P2P peer HTTP endpoints ──────────────────────────────────────

    /// /peer/health (handshake surface) stays reachable from loopback with no
    /// key — localhost broker auto-discovery relies on it. The CREDIT surface
    /// is covered separately by test_peer_credit_surface_fail_closed_without_key.
    #[test]
    fn test_peer_health_accessible_without_key() {
        let url = start_broker(free_port());
        let resp = ureq::get(&format!("{}/peer/health", url))
            .config()
            .timeout_global(Some(Duration::from_secs(5)))
            .build()
            .call()
            .unwrap();
        assert_eq!(resp.status().as_u16(), 200);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert_eq!(body["status"], "healthy");
    }

    /// FAIL-CLOSED: with no peer key configured, the P2P credit surface
    /// (reserve/commit/cancel/earn/balance/workers/tasks-offer) must reject
    /// ALL callers — even loopback — instead of silently skipping auth.
    #[test]
    fn test_peer_credit_surface_fail_closed_without_key() {
        let url = start_broker(free_port()); // peer_key: None

        // POST credit endpoints — no X-Peer-Key header at all.
        for (path, body) in [
            (
                "/peer/reserve",
                serde_json::json!({"user_id": "u", "amount": 1.0, "request_id": "r"}),
            ),
            (
                "/peer/commit",
                serde_json::json!({"reservation_id": "x", "actual_cost": 1.0}),
            ),
            ("/peer/cancel", serde_json::json!({"reservation_id": "x"})),
            (
                "/peer/earn",
                serde_json::json!({"amount": 1.0, "duration_ms": 0.0, "worker_id": "w", "requesting_user": "u", "request_id": "e"}),
            ),
            (
                "/peer/tasks/offer",
                serde_json::json!({"task_id": "t", "payload_b64": "", "max_price_per_hour": 1.0, "timeout_secs": 1.0}),
            ),
        ] {
            let resp = ureq::post(&format!("{}{}", url, path))
                .header("Content-Type", "application/json")
                .config()
                .timeout_global(Some(Duration::from_secs(5)))
                .http_status_as_error(false)
                .build()
                .send_json(body)
                .unwrap();
            assert_eq!(
                resp.status().as_u16(),
                401,
                "POST {} must be 401 when no peer key configured",
                path
            );
        }

        // GET credit/info endpoints.
        for path in ["/peer/balance?user_id=u", "/peer/workers"] {
            let resp = ureq::get(&format!("{}{}", url, path))
                .config()
                .timeout_global(Some(Duration::from_secs(5)))
                .http_status_as_error(false)
                .build()
                .call()
                .unwrap();
            assert_eq!(
                resp.status().as_u16(),
                401,
                "GET {} must be 401 when no peer key configured",
                path
            );
        }
    }

    /// Even presenting some X-Peer-Key is rejected when the broker has no key
    /// configured (verify_peer_key denies on empty expected key).
    #[test]
    fn test_peer_credit_surface_rejects_any_key_when_none_configured() {
        let url = start_broker(free_port()); // peer_key: None
        let resp = peer_post_json(
            &format!("{}/peer/reserve", url),
            serde_json::json!({"user_id": "u", "amount": 1.0, "request_id": "r"}),
            "attacker-guessed-key",
        );
        assert_eq!(resp.status().as_u16(), 401);
    }

    /// /peer/health rejects wrong peer key and accepts correct one.
    #[test]
    fn test_peer_health_rejects_wrong_key() {
        let url = start_broker_p2p(free_port());

        let resp = peer_get_json(&format!("{}/peer/health", url), "wrong-key");
        assert_eq!(resp.status().as_u16(), 401);

        let resp = peer_get_json(&format!("{}/peer/health", url), "test-peer-key");
        assert_eq!(resp.status().as_u16(), 200);
    }

    /// /peers exposes mesh topology and must require the peer key when one is
    /// configured (audit M4: was an unauthenticated reconnaissance surface).
    #[test]
    fn test_peers_requires_key_when_configured() {
        let url = start_broker_p2p(free_port()); // peer_key = "test-peer-key"
                                                 // No key → 401.
        let resp = ureq::get(&format!("{}/peers", url))
            .config()
            .timeout_global(Some(Duration::from_secs(5)))
            .http_status_as_error(false)
            .build()
            .call()
            .unwrap();
        assert_eq!(
            resp.status().as_u16(),
            401,
            "/peers must 401 without the peer key"
        );
        // Correct key → 200.
        let ok = peer_get_json(&format!("{}/peers", url), "test-peer-key");
        assert_eq!(ok.status().as_u16(), 200);
    }

    /// Two-phase: an offer with two_phase=true RESERVES (returns TaskAccept) and
    /// does NOT execute; cancel releases it.
    #[test]
    fn test_two_phase_offer_reserves_without_executing() {
        let url = start_broker_p2p(free_port());
        register_worker_at(&url, "local-w", "http://127.0.0.1:3960", 1.0);
        let offer = serde_json::json!({
            "task_id": "tp-1", "payload_b64": "", "max_price_per_hour": 100.0,
            "estimated_duration_secs": 1.0, "timeout_secs": 10.0,
            "requester_user_id": "u", "source_broker": "n", "two_phase": true
        });
        let resp = peer_post_json(&format!("{}/peer/tasks/offer", url), offer, "test-peer-key");
        assert_eq!(resp.status().as_u16(), 200);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        // A TaskAccept (reserve) — has estimated_cost; NOT a TaskResult (no payload_b64).
        assert!(
            body.get("estimated_cost").is_some(),
            "expected TaskAccept, got {:?}",
            body
        );
        assert!(
            body.get("payload_b64").is_none(),
            "must not have executed the task"
        );
        // Cancel releases the reservation.
        let c = peer_post_json(
            &format!("{}/peer/tasks/cancel", url),
            serde_json::json!({"task_id": "tp-1"}),
            "test-peer-key",
        );
        assert_eq!(c.status().as_u16(), 200);
    }

    /// Two-phase: committing an unknown/expired task_id is rejected (404).
    #[test]
    fn test_two_phase_commit_unknown_rejects() {
        let url = start_broker_p2p(free_port());
        let resp = peer_post_json(
            &format!("{}/peer/tasks/commit", url),
            serde_json::json!({"task_id": "nope"}),
            "test-peer-key",
        );
        assert_eq!(resp.status().as_u16(), 404);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert_eq!(body["task_id"], "nope");
    }

    /// The two-phase commit/cancel endpoints require the peer key (M1 fail-closed).
    #[test]
    fn test_two_phase_endpoints_require_peer_key() {
        let url = start_broker_p2p(free_port());
        let resp = ureq::post(&format!("{}/peer/tasks/commit", url))
            .header("Content-Type", "application/json")
            .config()
            .timeout_global(Some(Duration::from_secs(5)))
            .http_status_as_error(false)
            .build()
            .send_json(serde_json::json!({"task_id": "x"}))
            .unwrap();
        assert_eq!(resp.status().as_u16(), 401);
    }

    /// Full P2P lifecycle: earn → reserve → commit → verify balance.
    #[test]
    fn test_peer_reserve_commit_cycle_via_http() {
        let url = start_broker_p2p(free_port());
        let pk = "test-peer-key";

        // Seed 100 credits via /peer/earn (adds to owner's authoritative balance)
        let resp = peer_post_json(
            &format!("{}/peer/earn", url),
            serde_json::json!({
                "amount": 100.0,
                "duration_ms": 1000.0,
                "worker_id": "test-w",
                "requesting_user": "someone",
                "request_id": "earn-1"
            }),
            pk,
        );
        assert_eq!(resp.status().as_u16(), 200);

        // Verify balance
        let resp = peer_get_json(&format!("{}/peer/balance?user_id=test-owner", url), pk);
        assert_eq!(resp.status().as_u16(), 200);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert!((body["balance"].as_f64().unwrap() - 100.0).abs() < 1e-4);

        // Reserve 40
        let resp = peer_post_json(
            &format!("{}/peer/reserve", url),
            serde_json::json!({"user_id": "test-owner", "amount": 40.0, "request_id": "res-1"}),
            pk,
        );
        assert_eq!(resp.status().as_u16(), 200);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        let reservation_id = body["reservation_id"].as_str().unwrap().to_string();
        assert!((body["balance_before"].as_f64().unwrap() - 100.0).abs() < 1e-4);

        // Balance should be 60
        let resp = peer_get_json(&format!("{}/peer/balance?user_id=test-owner", url), pk);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert!((body["balance"].as_f64().unwrap() - 60.0).abs() < 1e-4);

        // Commit with actual_cost 25 → refund 15 → balance 75
        let resp = peer_post_json(
            &format!("{}/peer/commit", url),
            serde_json::json!({"reservation_id": reservation_id, "actual_cost": 25.0}),
            pk,
        );
        assert_eq!(resp.status().as_u16(), 200);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert!((body["balance_after"].as_f64().unwrap() - 75.0).abs() < 1e-4);
    }

    /// /peer/cancel fully refunds a reservation.
    #[test]
    fn test_peer_cancel_refunds_via_http() {
        let url = start_broker_p2p(free_port());
        let pk = "test-peer-key";

        // Seed
        peer_post_json(
            &format!("{}/peer/earn", url),
            serde_json::json!({"amount": 50.0, "duration_ms": 0.0, "worker_id": "w", "requesting_user": "u", "request_id": "e1"}),
            pk,
        );

        // Reserve 30
        let resp = peer_post_json(
            &format!("{}/peer/reserve", url),
            serde_json::json!({"user_id": "test-owner", "amount": 30.0, "request_id": "r2"}),
            pk,
        );
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        let rid = body["reservation_id"].as_str().unwrap().to_string();

        // Cancel → full refund
        let resp = peer_post_json(
            &format!("{}/peer/cancel", url),
            serde_json::json!({"reservation_id": rid}),
            pk,
        );
        assert_eq!(resp.status().as_u16(), 200);

        // Balance back to 50
        let resp = peer_get_json(&format!("{}/peer/balance?user_id=test-owner", url), pk);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert!((body["balance"].as_f64().unwrap() - 50.0).abs() < 1e-4);
    }

    /// /peer/reserve returns 402 when user has insufficient credits.
    #[test]
    fn test_peer_reserve_insufficient_returns_402() {
        let url = start_broker_p2p(free_port());
        let pk = "test-peer-key";

        let resp = peer_post_json(
            &format!("{}/peer/reserve", url),
            serde_json::json!({"user_id": "test-owner", "amount": 100.0, "request_id": "r3"}),
            pk,
        );
        assert_eq!(resp.status().as_u16(), 402);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert_eq!(body["code"], "INSUFFICIENT_CREDITS");
    }

    // ── Transaction buffer ───────────────────────────────────────────

    /// Transactions are queued and counted correctly.
    #[test]
    fn test_transaction_buffer_push_and_drain() {
        use crate::broker::flush::{BufferedTransaction, TransactionBuffer};
        let buf = TransactionBuffer::new();
        assert_eq!(buf.pending_count(), 0);
        assert_eq!(buf.total_flushed(), 0);

        buf.push_transaction(BufferedTransaction {
            request_id: "tx-1".to_string(),
            user_id: "alice".to_string(),
            tx_type: "commit".to_string(),
            amount: 0.05,
            balance_after: 99.95,
            worker_id: "w-1".to_string(),
            duration_ms: 150.0,
            source_node: Some("node-1".to_string()),
            worker_name: Some("worker-1".to_string()),
            worker_uri: Some("http://10.13.13.5:3960".to_string()),
            price_per_hour: 3.6,
        });
        buf.push_transaction(BufferedTransaction {
            request_id: "tx-2".to_string(),
            user_id: "alice".to_string(),
            tx_type: "commit".to_string(),
            amount: 0.10,
            balance_after: 99.85,
            worker_id: "w-1".to_string(),
            duration_ms: 200.0,
            source_node: None,
            worker_name: None,
            worker_uri: None,
            price_per_hour: 0.0,
        });

        assert_eq!(buf.pending_count(), 2);
    }

    /// Balance snapshots keep only the latest value per user (DashMap insert overwrites).
    #[test]
    fn test_transaction_buffer_balance_snapshots() {
        use crate::broker::flush::TransactionBuffer;
        let buf = TransactionBuffer::new();

        buf.snapshot_balance("alice", 100.0);
        buf.snapshot_balance("alice", 95.0);
        buf.snapshot_balance("bob", 200.0);

        // Internal state: alice → 95.0, bob → 200.0
        // No pending transactions
        assert_eq!(buf.pending_count(), 0);
    }

    /// GET /credits/:user returns balance_status field for dual-mode visibility.
    #[test]
    fn test_credits_endpoint_returns_balance_status() {
        let url = start_broker(free_port());
        add_credits(&url, "status-user", 50.0);

        let resp = ureq::get(&format!("{}/credits/status-user", url))
            .header("Authorization", "Bearer zk_status-user_test")
            .config()
            .timeout_global(Some(Duration::from_secs(5)))
            .build()
            .call()
            .unwrap();
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert_eq!(body["balance_status"], "authoritative");
        assert!(body["last_prefetched"].is_null());
    }

    // ═══════════════════════════════════════════════════════════════════
    // PILLAR 2 — LOCAL vs REMOTE CLUSTER ACCESS
    // ═══════════════════════════════════════════════════════════════════

    /// In local mode, execute through a real mock worker succeeds with zero cost.
    #[test]
    fn test_local_mode_mock_worker_execute_no_credits_charged() {
        let (worker_url, _) = start_mock_worker();
        let broker_url = start_broker(free_port());

        add_credits(&broker_url, "local-user", 100.0);
        let balance_before = get_balance(&broker_url, "local-user");
        assert!((balance_before - 100.0).abs() < 0.001);

        // Register mock worker (localhost URI → local = free)
        register_worker_at(&broker_url, "mock-w", &worker_url, 3.6);

        // Execute (local mode: X-Zakuro-User trusted, no auth needed)
        let resp = post_json_with(
            &format!("{}/execute", broker_url),
            serde_json::json!({"fn": "test_fn", "args": []}),
            &[("X-Zakuro-User", "local-user")],
        );
        assert_eq!(resp.status().as_u16(), 200);

        // Balance unchanged (local = free execution)
        let balance_after = get_balance(&broker_url, "local-user");
        assert!(
            (balance_after - 100.0).abs() < 0.001,
            "balance should be unchanged in local mode, got {}",
            balance_after
        );
    }

    /// POST /execute round-trips through the async axum route (Task 6 — Phase 2
    /// outbound).  The async route is the ONLY handler for POST /execute; this
    /// test confirms the worker response body and 200 status are preserved.
    #[test]
    fn test_async_execute_route_round_trips_worker_response() {
        let (worker_url, _) = start_mock_worker();
        let broker_url = start_broker(free_port());

        // Register mock worker (localhost URI → local = free, no credit deduction).
        register_worker_at(&broker_url, "async-mock-w", &worker_url, 3.6);

        // POST /execute — goes through the async axum route.
        let resp = ureq::post(&format!("{}/execute", broker_url))
            .header("Content-Type", "application/json")
            .header("X-Zakuro-User", "async-test-user")
            .config()
            .timeout_global(Some(std::time::Duration::from_secs(10)))
            .http_status_as_error(false)
            .build()
            .send_json(serde_json::json!({"fn": "test_fn", "args": []}))
            .unwrap();

        assert_eq!(
            resp.status().as_u16(),
            200,
            "async /execute route must return 200"
        );

        let body = resp.into_body().read_to_string().unwrap();
        assert_eq!(
            body, "mock-result",
            "async /execute route must return the worker's body verbatim"
        );
    }

    /// Remote mode (discovery disabled → local_mode stays false) requires Bearer auth.
    #[test]
    fn test_remote_mode_requires_bearer_auth() {
        let url = start_broker_remote(free_port());

        let resp = post_json_err(
            &format!("{}/execute", url),
            serde_json::json!({"fn": "test"}),
        );
        assert_eq!(resp.status().as_u16(), 401);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert_eq!(body["code"], "UNAUTHORIZED");
    }

    /// Remote mode rejects execute when user has no credits for a non-local worker.
    #[test]
    fn test_remote_mode_enforces_credits_for_non_local_worker() {
        // ── Without API credentials: billing disabled, no credit enforcement ──
        // Use a port that refuses immediately (no TCP timeout) but NOT 127.0.0.1
        // (which is always considered "local").
        let url_free = start_broker_remote(free_port());
        register_worker_at(&url_free, "remote-w", "http://10.99.99.1:3960", 3.6);
        let status_free = ureq::post(&format!("{}/execute", url_free))
            .header("Authorization", "Bearer zk_nocreds_abc123def")
            .config()
            .timeout_global(Some(Duration::from_secs(5)))
            .http_status_as_error(false)
            .build()
            .send_json(serde_json::json!({"fn": "test"}))
            .map(|r| r.status().as_u16())
            .unwrap_or(0); // transport / timeout — still not 402
        assert_ne!(
            status_free, 402,
            "no API credentials → billing disabled → must never get 402"
        );

        // ── With API credentials: billing enforced, insufficient credits → 402 ──
        let port_billed = free_port();
        let cfg = BrokerConfig {
            enable_discovery: false,
            api_url: Some("http://test-authority".to_string()),
            api_key: Some("test-key".to_string()),
            ..test_config(port_billed)
        };
        thread::spawn(move || {
            let _ = start_server(cfg);
        });
        let url_billed = format!("http://127.0.0.1:{}", port_billed);
        for _ in 0..300 {
            if ureq::get(&format!("{}/health", url_billed))
                .config()
                .timeout_global(Some(Duration::from_millis(150)))
                .build()
                .call()
                .is_ok()
            {
                break;
            }
            thread::sleep(Duration::from_millis(100));
        }
        register_worker_at(&url_billed, "remote-w", "http://10.99.99.1:3960", 3.6);
        let resp = post_json_with(
            &format!("{}/execute", url_billed),
            serde_json::json!({"fn": "test"}),
            &[("Authorization", "Bearer zk_nocreds_abc123def")],
        );
        assert_eq!(
            resp.status().as_u16(),
            402,
            "billing enabled → must get 402 for zero credits"
        );
    }

    /// P2P publish-lock path must reject a zero-credit user with 402 BEFORE
    /// dispatching the task to any peer (mirrors the local select_worker gate).
    /// Regression for audit C2: dispatch had no pre-execution balance check.
    #[test]
    fn test_p2p_dispatch_rejects_insufficient_credits_before_dispatch() {
        // Discovery OFF → local_mode stays false; billing ON → credit checks run;
        // P2P ON → execute takes the publish-lock branch. No peers registered.
        let port = free_port();
        let cfg = BrokerConfig {
            enable_discovery: false,
            enable_p2p: true,
            peer_key: Some("test-peer-key".to_string()),
            owner_user_id: Some("test-owner".to_string()),
            api_url: Some("http://test-authority".to_string()),
            api_key: Some("test-key".to_string()),
            ..test_config(port)
        };
        thread::spawn(move || {
            let _ = start_server(cfg);
        });
        let url = format!("http://127.0.0.1:{}", port);
        for _ in 0..300 {
            if ureq::get(&format!("{}/health", url))
                .config()
                .timeout_global(Some(Duration::from_millis(150)))
                .build()
                .call()
                .is_ok()
            {
                break;
            }
            thread::sleep(Duration::from_millis(100));
        }

        // No local worker + remote_only forces the dispatch_to_peers path.
        // User "zk_nocreds_..." has zero balance → must be rejected pre-dispatch.
        let resp = post_json_with(
            &format!("{}/execute", url),
            serde_json::json!({"fn": "test"}),
            &[
                ("Authorization", "Bearer zk_nocreds_abc123def"),
                (
                    "X-Zakuro-Requirements",
                    r#"{"remote_only": true, "estimated_duration_secs": 60.0, "budget_credits": 1.0}"#,
                ),
            ],
        );
        assert_eq!(
            resp.status().as_u16(),
            402,
            "P2P dispatch with zero credits must return 402, not dispatch the task"
        );
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert_eq!(body["code"], "INSUFFICIENT_CREDITS");
    }

    /// is_local_worker correctly identifies localhost, remote, and local-mode overrides.
    #[test]
    fn test_is_local_worker_comprehensive() {
        use crate::broker::BrokerState;
        let state = BrokerState::new();

        // Localhost variants → always local
        assert!(state.is_local_worker("http://127.0.0.1:3960"));
        assert!(state.is_local_worker("http://127.0.0.1:8089"));
        assert!(state.is_local_worker("http://localhost:3960"));
        assert!(state.is_local_worker("http://localhost:9000/execute"));

        // Remote IPs → not local (in default mode)
        assert!(!state.is_local_worker("http://10.13.13.5:3960"));
        assert!(!state.is_local_worker("http://100.64.0.1:3960"));
        assert!(!state.is_local_worker("http://192.168.1.100:3960"));

        // In local_mode, everything is local (even remote IPs)
        state.set_local_mode(true);
        assert!(state.is_local_worker("http://10.13.13.5:3960"));
        assert!(state.is_local_worker("http://100.64.0.1:3960"));

        // Reset and verify remote IPs are non-local again
        state.set_local_mode(false);
        assert!(!state.is_local_worker("http://10.13.13.5:3960"));
    }

    /// /peer/workers exposes this broker's OWN workers — including one
    /// registered with a reachable non-loopback address (`zc serve
    /// --advertise`, zc#172) — but never re-exports a peer-synced worker
    /// (source_node set), which would echo forever between brokers.
    #[test]
    fn test_peer_workers_exports_own_workers_never_peer_synced_ones() {
        let url = start_broker_p2p(free_port());
        let pk = "test-peer-key";

        register_worker_at(&url, "local-w", "http://127.0.0.1:3960", 1.0);
        // Own worker on a reachable LAN address — the advertise case. The old
        // loopback filter dropped this one, so an advertised provider was
        // invisible to the rest of the fleet.
        register_worker_at(&url, "advertised-w", "http://10.13.13.5:3960", 2.0);
        // A peer-synced worker carries its owning node; must not be re-exported.
        post_json(
            &format!("{}/workers", url),
            serde_json::json!({
                "name": "peer-w",
                "uri": "http://10.13.13.9:3960",
                "worker_type": "zakuro",
                "source_node": "zc://node-deadbeef"
            }),
        );

        // /workers returns all three
        let body = get_json(&format!("{}/workers", url));
        assert_eq!(body["total"], 3);

        let resp = peer_get_json(&format!("{}/peer/workers", url), pk);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert_eq!(body["total"], 2);
        let names: Vec<&str> = body["workers"]
            .as_array()
            .unwrap()
            .iter()
            .map(|w| w["name"].as_str().unwrap())
            .collect();
        assert!(names.contains(&"local-w") && names.contains(&"advertised-w"));
        assert!(!names.contains(&"peer-w"));
    }

    // ═══════════════════════════════════════════════════════════════════
    // PILLAR 3 — RELIABILITY: RETRY, PREFETCH, RECONCILIATION, PROOF
    // ═══════════════════════════════════════════════════════════════════

    // ── WAL proof of execution ───────────────────────────────────────

    /// Full WAL lifecycle: Reserved → Executed → Committed.
    /// Committed entries are excluded from read_uncommitted.
    #[test]
    fn test_wal_proof_of_execution_full_lifecycle() {
        use crate::broker::wal::{Wal, WalEntry, WalStatus};
        let path = format!("/tmp/zc_test_wal_proof_{}.jsonl", free_port());
        let _ = std::fs::remove_file(&path);

        let wal = Wal::open(&path).unwrap();
        wal.append(&WalEntry {
            request_id: "proof-1".to_string(),
            user_id: "alice".to_string(),
            reservation_id: "res-proof-1".to_string(),
            estimated_cost: 0.05,
            actual_cost: None,
            worker_id: "worker-a".to_string(),
            duration_ms: None,
            timestamp: chrono::Utc::now(),
            status: WalStatus::Reserved,
        })
        .unwrap();
        wal.flush_buffer().unwrap();

        let uncommitted = wal.read_uncommitted().unwrap();
        assert!(uncommitted
            .iter()
            .any(|e| e.request_id == "proof-1" && e.status == WalStatus::Reserved));

        wal.update_status("proof-1", WalStatus::Executed, Some(0.03), Some(250.0))
            .unwrap();
        wal.flush_buffer().unwrap();
        let uncommitted = wal.read_uncommitted().unwrap();
        assert!(uncommitted
            .iter()
            .any(|e| e.request_id == "proof-1" && e.status == WalStatus::Executed));

        wal.update_status("proof-1", WalStatus::Committed, Some(0.03), Some(250.0))
            .unwrap();
        wal.flush_buffer().unwrap();
        let uncommitted = wal.read_uncommitted().unwrap();
        assert!(
            !uncommitted.iter().any(|e| e.request_id == "proof-1"),
            "committed entry must not appear in uncommitted"
        );

        let _ = std::fs::remove_file(&path);
    }

    /// WAL recovery: stale Reserved entries are cancelled and marked Failed.
    #[test]
    fn test_wal_recovery_cancels_stale_reservations() {
        use crate::broker::ledger::Ledger;
        use crate::broker::recovery;
        use crate::broker::wal::{Wal, WalEntry, WalStatus};

        let path = format!("/tmp/zc_test_wal_cancel_{}.jsonl", free_port());
        let _ = std::fs::remove_file(&path);

        let ledger = Ledger::new(None, None);

        // Simulate a crash: Reserved entry left in WAL
        let wal = Wal::open(&path).unwrap();
        wal.append(&WalEntry {
            request_id: "stale-req".to_string(),
            user_id: "alice".to_string(),
            reservation_id: "res-stale".to_string(),
            estimated_cost: 30.0,
            actual_cost: None,
            worker_id: "w1".to_string(),
            duration_ms: None,
            timestamp: chrono::Utc::now(),
            status: WalStatus::Reserved,
        })
        .unwrap();
        wal.flush_buffer().unwrap();

        // Replay → stale reservation is cancelled (marked Failed)
        let tx_buffer = crate::broker::flush::TransactionBuffer::new();
        recovery::replay_wal(&wal, &ledger, &tx_buffer, None);

        let uncommitted = wal.read_uncommitted().unwrap();
        assert!(
            uncommitted.is_empty(),
            "all entries should be resolved after replay"
        );

        let _ = std::fs::remove_file(&path);
    }

    /// WAL recovery: Executed entries are committed (proof of work preserved).
    #[test]
    fn test_wal_recovery_commits_executed_entries() {
        use crate::broker::ledger::Ledger;
        use crate::broker::recovery;
        use crate::broker::wal::{Wal, WalEntry, WalStatus};

        let path = format!("/tmp/zc_test_wal_commit_{}.jsonl", free_port());
        let _ = std::fs::remove_file(&path);

        let ledger = Ledger::new(None, None);

        // Simulate crash after execution but before commit
        let wal = Wal::open(&path).unwrap();
        wal.append(&WalEntry {
            request_id: "exec-req".to_string(),
            user_id: "bob".to_string(),
            reservation_id: "res-exec".to_string(),
            estimated_cost: 0.03,
            actual_cost: Some(0.03),
            worker_id: "w2".to_string(),
            duration_ms: Some(150.0),
            timestamp: chrono::Utc::now(),
            status: WalStatus::Executed,
        })
        .unwrap();
        wal.flush_buffer().unwrap();

        let tx_buffer = crate::broker::flush::TransactionBuffer::new();
        recovery::replay_wal(&wal, &ledger, &tx_buffer, None);

        let uncommitted = wal.read_uncommitted().unwrap();
        assert!(
            uncommitted.is_empty(),
            "executed entry should be committed after replay"
        );

        let _ = std::fs::remove_file(&path);
    }

    /// MONEY-CRITICAL: an earn that was credited AND already confirmed
    /// flushed to the dashboard (WAL marked Committed) must NOT be re-applied
    /// on replay — that would mint credits out of thin air.
    #[test]
    fn test_wal_recovery_does_not_double_credit_committed_earn() {
        use crate::broker::flush::TransactionBuffer;
        use crate::broker::ledger::Ledger;
        use crate::broker::recovery;
        use crate::broker::wal::{Wal, WalEntry, WalStatus};

        let path = format!("/tmp/zc_test_wal_earn_committed_{}.jsonl", free_port());
        let _ = std::fs::remove_file(&path);

        let ledger = Ledger::new(None, None);

        let wal = Wal::open(&path).unwrap();
        wal.append(&WalEntry {
            request_id: "earn-req-committed".to_string(),
            user_id: "owner1".to_string(),
            reservation_id: String::new(),
            estimated_cost: 0.05,
            actual_cost: Some(0.05),
            worker_id: "worker-1".to_string(),
            duration_ms: Some(1000.0),
            timestamp: chrono::Utc::now(),
            status: WalStatus::Earned,
        })
        .unwrap();

        // Simulate what handle_peer_earn does: credit in-memory once.
        ledger.local_add_credits("owner1", 0.05);

        // Simulate what flush_to_api does on a confirmed dashboard delivery:
        // mark the WAL entry Committed. From here on it must be replay-immune.
        wal.update_status("earn-req-committed", WalStatus::Committed, Some(0.05), None)
            .unwrap();
        wal.flush_buffer().unwrap();

        // "Restart": replay must be a no-op for this entry since it's Committed.
        let tx_buffer = TransactionBuffer::new();
        recovery::replay_wal(&wal, &ledger, &tx_buffer, None);

        let balance = ledger
            .authoritative_balances
            .get("owner1")
            .map(|v| *v)
            .unwrap_or(0.0);
        assert!(
            (balance - 0.05).abs() < 1e-9,
            "earn must be applied exactly once, got balance {}",
            balance
        );
        assert_eq!(
            tx_buffer.pending_count(),
            0,
            "a committed earn must not be re-queued for delivery"
        );

        let _ = std::fs::remove_file(&path);
    }

    /// MONEY-CRITICAL: an earn that was credited in-memory but NEVER confirmed
    /// flushed (still `Earned`, simulating a crash before/mid-flush) must be
    /// replayed on restart — the payout must not be lost.
    #[test]
    fn test_wal_recovery_replays_unflushed_earn() {
        use crate::broker::flush::TransactionBuffer;
        use crate::broker::ledger::Ledger;
        use crate::broker::recovery;
        use crate::broker::wal::{Wal, WalEntry, WalStatus};

        let path = format!("/tmp/zc_test_wal_earn_uncommitted_{}.jsonl", free_port());
        let _ = std::fs::remove_file(&path);

        // Fresh ledger simulates the post-crash restart: in-memory state is gone.
        let ledger = Ledger::new(None, None);

        let wal = Wal::open(&path).unwrap();
        wal.append(&WalEntry {
            request_id: "earn-req-lost".to_string(),
            user_id: "owner2".to_string(),
            reservation_id: String::new(),
            estimated_cost: 0.07,
            actual_cost: Some(0.07),
            worker_id: "worker-2".to_string(),
            duration_ms: Some(2000.0),
            timestamp: chrono::Utc::now(),
            status: WalStatus::Earned,
        })
        .unwrap();
        wal.flush_buffer().unwrap();
        // Note: status is NOT marked Committed — this is the crash-before-flush case.

        let tx_buffer = TransactionBuffer::new();
        recovery::replay_wal(&wal, &ledger, &tx_buffer, None);

        // Replay must NOT touch the in-memory spendable balance: the dashboard
        // is authoritative and re-seeds authoritative_balances lazily. Adding
        // here would double-count against the dashboard's own copy of the earn.
        assert!(
            ledger.authoritative_balances.get("owner2").is_none(),
            "replay must not mutate in-memory balance (dashboard is authoritative)"
        );

        // It DOES re-buffer the earn for dashboard re-delivery (idempotent via
        // dashboard request_id dedup).
        assert_eq!(
            tx_buffer.pending_count(),
            1,
            "unflushed earn must be re-queued for dashboard delivery after replay"
        );

        // The WAL entry must still be replay-eligible (Earned) until the
        // re-queued delivery is actually confirmed flushed — replay itself
        // does not mark it Committed.
        let uncommitted = wal.read_uncommitted().unwrap();
        assert!(
            uncommitted.iter().any(|e| e.request_id == "earn-req-lost"),
            "replayed-but-not-yet-flushed earn must remain uncommitted"
        );

        // Simulate a successful flush confirming the earn (applied/duplicate):
        // the WAL entry transitions to Committed and is no longer replay-eligible.
        wal.update_status(
            "earn-req-lost",
            WalStatus::Committed,
            Some(0.07),
            Some(2000.0),
        )
        .unwrap();
        wal.flush_buffer().unwrap();
        let uncommitted = wal.read_uncommitted().unwrap();
        assert!(
            !uncommitted.iter().any(|e| e.request_id == "earn-req-lost"),
            "confirmed earn must no longer be replay-eligible"
        );

        let _ = std::fs::remove_file(&path);
    }

    /// MONEY-CRITICAL: flush honors per-item results[] — only applied/duplicate
    /// items are marked WAL-Committed; failed items are re-buffered and left Earned.
    #[test]
    fn test_flush_per_item_results_marks_only_durable_committed() {
        use crate::broker::flush::{BufferedTransaction, TransactionBuffer};
        use crate::broker::wal::{Wal, WalEntry, WalStatus};

        let path = format!("/tmp/zc_test_flush_per_item_{}.jsonl", free_port());
        let _ = std::fs::remove_file(&path);
        let wal = Wal::open(&path).unwrap();

        // Three earn WAL entries + tx_buffer entries: applied, duplicate, failed.
        let mk = |rid: &str| WalEntry {
            request_id: rid.to_string(),
            user_id: "owner3".to_string(),
            reservation_id: String::new(),
            estimated_cost: 1.0,
            actual_cost: Some(1.0),
            worker_id: "w".to_string(),
            duration_ms: Some(10.0),
            timestamp: chrono::Utc::now(),
            status: WalStatus::Earned,
        };
        for rid in ["earn-a", "earn-b", "earn-c"] {
            wal.append(&mk(rid)).unwrap();
        }
        wal.flush_buffer().unwrap();

        let buf = TransactionBuffer::new();
        for rid in ["earn-a", "earn-b", "earn-c"] {
            buf.push_transaction(BufferedTransaction {
                request_id: rid.to_string(),
                user_id: "owner3".to_string(),
                tx_type: "credit".to_string(),
                amount: 1.0,
                balance_after: 0.0,
                worker_id: "w".to_string(),
                duration_ms: 10.0,
                source_node: None,
                worker_name: Some("w".to_string()),
                worker_uri: None,
                price_per_hour: 0.0,
            });
        }

        // Spin up a mock dashboard that returns mixed per-item results.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let handle = std::thread::spawn(move || {
            use std::io::{Read, Write};
            let (mut stream, _) = listener.accept().unwrap();
            let mut buf = [0u8; 8192];
            let _ = stream.read(&mut buf);
            let body = r#"{"results":[
                {"request_id":"earn-a","status":"applied","error":null},
                {"request_id":"earn-b","status":"duplicate","error":null},
                {"request_id":"earn-c","status":"failed","error":"boom"}],
                "applied":1,"duplicates":1,"failed":1,"success":false,
                "inserted":2,"errors":[{"job_name":"x","error":"boom"}]}"#;
            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
            );
            let _ = stream.write_all(resp.as_bytes());
            let _ = stream.flush();
        });

        let api_url = format!("http://{}", addr);
        buf.flush_to_api(&api_url, "test-key", &wal);
        handle.join().unwrap();

        wal.flush_buffer().unwrap();
        let uncommitted: std::collections::HashSet<String> = wal
            .read_uncommitted()
            .unwrap()
            .into_iter()
            .map(|e| e.request_id)
            .collect();

        // applied + duplicate → Committed (not uncommitted).
        assert!(
            !uncommitted.contains("earn-a"),
            "applied earn must be Committed"
        );
        assert!(
            !uncommitted.contains("earn-b"),
            "duplicate earn must be Committed"
        );
        // failed → still Earned (uncommitted) AND re-buffered for retry.
        assert!(
            uncommitted.contains("earn-c"),
            "failed earn must remain Earned/replay-eligible"
        );
        assert_eq!(
            buf.pending_count(),
            1,
            "failed item must be re-buffered for retry"
        );

        let _ = std::fs::remove_file(&path);
    }

    // ── Prefetch & Reconciliation ────────────────────────────────────

    /// Balance status transitions: Authoritative → Prefetched → Reconciling → Authoritative.
    #[test]
    fn test_prefetched_balance_status_transitions() {
        use crate::broker::credits::{BalanceStatus, CreditManager};
        let mgr = CreditManager::new();

        mgr.get_or_create("peer-user", 100.0);
        assert_eq!(
            mgr.get("peer-user").unwrap().balance_status,
            BalanceStatus::Authoritative
        );

        mgr.set_prefetched_balance("peer-user", 95.0);
        let uc = mgr.get("peer-user").unwrap();
        assert_eq!(uc.balance_status, BalanceStatus::Prefetched);
        assert_eq!(uc.balance, 95.0);
        assert!(uc.last_prefetched.is_some());

        mgr.set_reconciling("peer-user");
        assert_eq!(
            mgr.get("peer-user").unwrap().balance_status,
            BalanceStatus::Reconciling
        );

        mgr.set_authoritative("peer-user");
        let uc = mgr.get("peer-user").unwrap();
        assert_eq!(uc.balance_status, BalanceStatus::Authoritative);
        assert!(uc.last_prefetched.is_none());
    }

    /// needs_reconciliation detects stale prefetched balances.
    #[test]
    fn test_reconciliation_needed_after_staleness() {
        use crate::broker::credits::CreditManager;
        let mgr = CreditManager::new();
        mgr.get_or_create("stale-user", 100.0);

        // Authoritative → never needs reconciliation
        assert!(!mgr.needs_reconciliation("stale-user", 0));

        mgr.set_prefetched_balance("stale-user", 90.0);

        // max_age_secs=0 → any prefetched balance is stale
        assert!(mgr.needs_reconciliation("stale-user", 0));

        // max_age_secs=999999 → recently prefetched, not stale yet
        assert!(!mgr.needs_reconciliation("stale-user", 999999));

        // Unknown user → no reconciliation needed
        assert!(!mgr.needs_reconciliation("ghost-user", 0));
    }

    // ── Retry policy ─────────────────────────────────────────────────

    /// Failed worker forward marks the worker unhealthy and triggers retry.
    #[test]
    fn test_retry_marks_unreachable_workers_unhealthy() {
        let url = start_broker(free_port());

        register_worker_at(&url, "unreachable-1", "http://127.0.0.1:19991", 0.001);
        register_worker_at(&url, "unreachable-2", "http://127.0.0.1:19992", 0.001);

        // Verify both are healthy initially
        let workers = get_json(&format!("{}/workers", url));
        assert_eq!(workers["total"], 2);
        for w in workers["workers"].as_array().unwrap() {
            assert_eq!(w["status"], "healthy");
        }

        // Execute → forward fails → retry → also fails → both marked unhealthy
        let resp = post_json_err(
            &format!("{}/execute", url),
            serde_json::json!({"fn": "unreachable_test"}),
        );
        assert_eq!(resp.status().as_u16(), 503);

        let workers = get_json(&format!("{}/workers", url));
        let unhealthy_count = workers["workers"]
            .as_array()
            .unwrap()
            .iter()
            .filter(|w| w["status"] == "unhealthy")
            .count();
        assert_eq!(
            unhealthy_count, 2,
            "both unreachable workers should be marked unhealthy after retry"
        );
    }

    /// Retry exhausts all workers (MAX_RETRIES=2), returns 503 with descriptive error.
    #[test]
    fn test_retry_exhausts_all_workers_returns_503() {
        let url = start_broker(free_port());

        // 3 unreachable workers: 1 initial + 2 retries = 3 attempts
        register_worker_at(&url, "dead-1", "http://127.0.0.1:19993", 0.001);
        register_worker_at(&url, "dead-2", "http://127.0.0.1:19994", 0.001);
        register_worker_at(&url, "dead-3", "http://127.0.0.1:19995", 0.001);

        let resp = post_json_err(
            &format!("{}/execute", url),
            serde_json::json!({"fn": "fail_test"}),
        );
        assert_eq!(resp.status().as_u16(), 503);
        let body: serde_json::Value = resp.into_body().read_json().unwrap();
        assert_eq!(body["code"], "NO_WORKERS");
        let error_msg = body["error"].as_str().unwrap();
        assert!(
            error_msg.contains("unreachable") || error_msg.contains("retries"),
            "error should mention retry exhaustion: {}",
            error_msg
        );
    }

    /// After a failed execute with retries, a newly registered healthy worker is used.
    #[test]
    fn test_retry_failure_then_new_healthy_worker_succeeds() {
        let url = start_broker(free_port());

        register_worker_at(&url, "bad-w", "http://127.0.0.1:19996", 0.001);

        let resp = post_json_err(
            &format!("{}/execute", url),
            serde_json::json!({"fn": "fail"}),
        );
        assert_eq!(resp.status().as_u16(), 503);

        // Register a real mock worker
        let (mock_url, _) = start_mock_worker();
        register_worker_at(&url, "good-w", &mock_url, 0.001);

        let resp = post_json_err(
            &format!("{}/execute", url),
            serde_json::json!({"fn": "succeed"}),
        );
        assert_eq!(
            resp.status().as_u16(),
            200,
            "should succeed with healthy worker"
        );
    }

    // ── Multi-window rate limiting ───────────────────────────────────

    /// Per-second and per-day rate limits are enforced together.
    #[test]
    fn test_multi_window_rate_limits_enforced_together() {
        use crate::broker::credits::CreditManager;
        let mgr = CreditManager::new();
        mgr.get_or_create("limited-user", 1000.0);
        mgr.set_rate_limits("limited-user", Some(5), Some(10), None);
        mgr.set_rate_limit("limited-user", 1000);

        for _ in 0..5 {
            assert!(mgr.check_rate_limit("limited-user"));
        }
        // 6th blocked by per-second
        assert!(!mgr.check_rate_limit("limited-user"));
    }

    // ── Ledger local-fallback round-trip (API-mode fallback path) ────

    /// The full reserve → commit → balance cycle works through local in-memory ops.
    #[test]
    fn test_ledger_local_fallback_full_cycle() {
        use crate::broker::ledger::Ledger;
        let ledger = Ledger::new(None, None);

        // Seed credits directly
        ledger.local_credits.insert("user1".to_string(), 100.0);
        assert_eq!(ledger.get_balance("user1"), 100.0);

        let rid = ledger.reserve("user1", 40.0, "req-1").unwrap();
        assert_eq!(
            ledger.local_credits.get("user1").map(|v| *v).unwrap_or(0.0),
            60.0
        );

        let _balance = ledger.commit(&rid, 25.0).unwrap();
        // 60 + 15 refund = 75
        assert_eq!(
            ledger.local_credits.get("user1").map(|v| *v).unwrap_or(0.0),
            75.0
        );

        // Cancel non-existent reservation is idempotent
        assert!(ledger.cancel("non-existent").is_ok());
    }

    // ═══════════════════════════════════════════════════════════════════
    // API MODE — PG ACCESS IS FORBIDDEN
    // ═══════════════════════════════════════════════════════════════════

    /// When api_url and api_key are both set (API mode), is_api_mode() returns true.
    #[test]
    fn test_api_mode_skips_pg_pool_creation() {
        use crate::broker::ledger::Ledger;
        let ledger = Ledger::new(
            Some("https://my.zakuro-ai.com".to_string()),
            Some("zk_test_abc123".to_string()),
        );

        assert!(ledger.is_api_mode());
    }

    /// In API mode, the full credit cycle (reserve → commit → cancel)
    /// works entirely through local in-memory ops.
    #[test]
    fn test_api_mode_credit_cycle_uses_local_only() {
        use crate::broker::ledger::Ledger;
        let ledger = Ledger::new(
            Some("https://my.zakuro-ai.com".to_string()),
            Some("zk_test_abc123".to_string()),
        );

        assert!(ledger.is_api_mode());

        // Seed credits directly
        ledger.local_credits.insert("api-user".to_string(), 100.0);

        // Reserve
        let rid = ledger.reserve("api-user", 40.0, "req-1").unwrap();
        assert_eq!(
            ledger
                .local_credits
                .get("api-user")
                .map(|v| *v)
                .unwrap_or(0.0),
            60.0
        );

        // Commit with partial refund
        let _ = ledger.commit(&rid, 25.0).unwrap();
        assert_eq!(
            ledger
                .local_credits
                .get("api-user")
                .map(|v| *v)
                .unwrap_or(0.0),
            75.0
        );

        // Cancel non-existent reservation is idempotent
        assert!(ledger.cancel("non-existent").is_ok());
    }

    /// WAL recovery cancel_from_wal uses local in-memory ops.
    #[test]
    fn test_api_mode_wal_recovery_cancel_uses_local_fallback() {
        use crate::broker::ledger::Ledger;
        let ledger = Ledger::new(
            Some("https://my.zakuro-ai.com".to_string()),
            Some("zk_test_abc123".to_string()),
        );

        // cancel_from_wal should succeed (local in-memory)
        assert!(ledger.cancel_from_wal("user-x", 50.0).is_ok());
        // The refund should be reflected in local_credits
        let local_balance = ledger
            .local_credits
            .get("user-x")
            .map(|v| *v)
            .unwrap_or(0.0);
        assert_eq!(local_balance, 50.0);
    }

    /// WAL recovery commit_from_wal uses local in-memory ops.
    #[test]
    fn test_api_mode_wal_recovery_commit_uses_local_fallback() {
        use crate::broker::ledger::Ledger;
        let ledger = Ledger::new(
            Some("https://my.zakuro-ai.com".to_string()),
            Some("zk_test_abc123".to_string()),
        );

        // commit_from_wal with refund should succeed
        let balance = ledger.commit_from_wal("user-y", 0.10, 0.06).unwrap();
        // Refund = 0.10 - 0.06 = 0.04 → added to local_credits
        assert!((balance - 0.04).abs() < 1e-9);
    }

    /// In non-API mode (no api_url/api_key), is_api_mode() returns false.
    #[test]
    fn test_non_api_mode_attempts_pg_connection() {
        use crate::broker::ledger::Ledger;
        let ledger = Ledger::new(None, None);

        assert!(!ledger.is_api_mode());
    }

    // ═══════════════════════════════════════════════════════════════════
    // P2P MESH — 500 FIBONACCI CROSS-NODE BENCHMARK
    // ═══════════════════════════════════════════════════════════════════

    fn fibonacci(n: u64) -> u64 {
        if n <= 1 {
            return n;
        }
        let (mut a, mut b) = (0u64, 1u64);
        for _ in 2..=n {
            let c = a + b;
            a = b;
            b = c;
        }
        b
    }

    /// Start a fibonacci worker on the given IP (binds to ip:0, OS picks port).
    /// Returns (uri, port). The worker responds to POST /execute with {"result": fib(n)}.
    fn start_fibonacci_worker(bind_ip: &str) -> (String, u16) {
        let addr = format!("{}:0", bind_ip);
        let server = tiny_http::Server::http(&addr)
            .unwrap_or_else(|e| panic!("bind {}:0 failed: {}", bind_ip, e));
        let port = server.server_addr().to_ip().unwrap().port();
        thread::spawn(move || {
            for mut req in server.incoming_requests() {
                let mut buf = Vec::new();
                let _ = req.as_reader().read_to_end(&mut buf);
                let n = serde_json::from_slice::<serde_json::Value>(&buf)
                    .ok()
                    .and_then(|v| v["n"].as_u64())
                    .unwrap_or(4);
                let body = serde_json::json!({"result": fibonacci(n), "n": n});
                let _ = req.respond(
                    tiny_http::Response::from_data(serde_json::to_vec(&body).unwrap())
                        .with_status_code(200)
                        .with_header(
                            tiny_http::Header::from_bytes("Content-Type", "application/json")
                                .unwrap(),
                        ),
                );
            }
        });
        (format!("http://{}:{}", bind_ip, port), port)
    }

    /// Start a P2P broker bound to 0.0.0.0 so it is reachable on any loopback IP.
    /// Pass `api_url`/`api_key` to enable billing through the centralized API;
    /// leave them `None` for free broker-to-broker mode.
    fn start_mesh_broker(
        port: u16,
        own_ip: &str,
        owner: &str,
        peer_key: &str,
        peers: Vec<String>,
        api_url: Option<String>,
        api_key: Option<String>,
    ) -> String {
        start_mesh_broker_with_quic(port, own_ip, owner, peer_key, peers, api_url, api_key, None)
    }

    // A test fixture builder: every parameter is a knob a test sets to
    // describe the broker it wants. Bundling them into a struct would move the
    // same list one level away from the three call sites without making any of
    // them clearer, so the lint is allowed here rather than worked around.
    #[allow(clippy::too_many_arguments)]
    fn start_mesh_broker_with_quic(
        port: u16,
        own_ip: &str,
        owner: &str,
        peer_key: &str,
        peers: Vec<String>,
        api_url: Option<String>,
        api_key: Option<String>,
        quic_port: Option<u16>,
    ) -> String {
        let cfg = BrokerConfig {
            host: "0.0.0.0".to_string(),
            port,
            health_check_interval: 2,
            worker_timeout: 30,
            min_credits: 0.0001,
            daemon: true,
            verbose: false,
            tui_mode: false,
            enable_discovery: false,
            enable_p2p: true,
            peer_key: Some(peer_key.to_string()),
            owner_user_id: Some(owner.to_string()),
            node_name: Some(format!("mesh-{}", own_ip)),
            worker_key: None,
            api_url,
            api_key,
            wireguard_ip_override: Some(own_ip.to_string()),
            quic_port,
            runtime_worker_threads: None,
            discovery: DiscoveryConfig {
                subnet: "10.13.13".to_string(),
                worker_port: 3960,
                extra_ports: vec![],
                scan_port_range: None,
                interval_secs: 60,
                enable_scan: false,
                enable_dns: false,
                peers,
                local_workers: vec![],
            },
        };
        thread::spawn(move || {
            let _ = start_server(cfg);
        });
        let url = format!("http://127.0.0.1:{}", port);
        for _ in 0..300 {
            if ureq::get(&format!("{}/health", url))
                .config()
                .timeout_global(Some(Duration::from_millis(200)))
                .build()
                .call()
                .is_ok()
            {
                return url;
            }
            thread::sleep(Duration::from_millis(100));
        }
        panic!("Mesh broker on port {} did not start within 30s", port);
    }

    /// 500 fibonacci(4) requests across a 2-node P2P mesh — no currency.
    ///
    /// Topology:
    ///   Node01  broker=0.0.0.0:B1  own_ip=127.0.0.2  worker=127.0.0.2:W1
    ///   Node02  broker=0.0.0.0:B2  own_ip=127.0.0.3  worker=127.0.0.3:W2
    ///
    /// No API credentials → billing disabled.  All executions are free.
    /// The test verifies compute correctness, P2P routing, and throughput.
    #[test]
    fn test_p2p_mesh_fibonacci_500() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        use std::time::Instant;

        let peer_key = "fib-mesh-key";
        let users = ["alice", "bob", "carol", "dave"];
        let total_req = 500usize;
        let threads = 10usize;
        let per_thread = total_req / threads;

        // ── Workers ────────────────────────────────────────────────────
        let (w1_uri, w1_port) = start_fibonacci_worker("127.0.0.2");
        let (w2_uri, w2_port) = start_fibonacci_worker("127.0.0.3");

        // ── Brokers (no API → billing disabled) ─────────────────────────
        let b1_port = free_port();
        let b2_port = free_port();

        let b1_url = start_mesh_broker(
            b1_port,
            "127.0.0.2",
            "node01-owner",
            peer_key,
            vec![format!("http://127.0.0.3:{}", b2_port)],
            None,
            None,
        );
        let b2_url = start_mesh_broker(
            b2_port,
            "127.0.0.3",
            "node02-owner",
            peer_key,
            vec![format!("http://127.0.0.2:{}", b1_port)],
            None,
            None,
        );

        // ── Register workers on both brokers ───────────────────────────
        for base in [&b1_url, &b2_url] {
            register_worker_at(base, "fib-node01", &w1_uri, 0.001);
            register_worker_at(base, "fib-node02", &w2_uri, 0.001);
        }

        // No credit seeding — billing is disabled, everything is free.

        // ── Fire 500 requests ──────────────────────────────────────────
        struct Row {
            user: String,
            broker: &'static str,
            latency_ms: f64,
            ok: bool,
            fib_ok: bool,
        }

        let rows = Arc::new(std::sync::Mutex::new(Vec::<Row>::with_capacity(total_req)));
        let idx = Arc::new(AtomicUsize::new(0));
        let clock = Instant::now();

        let agent = ureq::Agent::new_with_config(
            ureq::Agent::config_builder()
                .timeout_global(Some(Duration::from_secs(30)))
                .http_status_as_error(false)
                .build(),
        );

        let mut handles = Vec::new();
        for _ in 0..threads {
            let rows = rows.clone();
            let idx = idx.clone();
            let agent = agent.clone();
            let b1 = b1_url.clone();
            let b2 = b2_url.clone();

            handles.push(thread::spawn(move || {
                for _ in 0..per_thread {
                    let i = idx.fetch_add(1, Ordering::SeqCst);
                    let user = users[i % users.len()];
                    let broker = if i.is_multiple_of(2) {
                        (&b1, "Node01")
                    } else {
                        (&b2, "Node02")
                    };
                    let key = format!("zk_{}_0001", user);

                    let t0 = Instant::now();
                    let resp = agent
                        .post(&format!("{}/execute", broker.0))
                        .header("Authorization", &format!("Bearer {}", key))
                        .header(
                            "X-Zakuro-Requirements",
                            r#"{"strategy":"round_robin","estimated_duration_secs":0.01}"#,
                        )
                        .send_json(serde_json::json!({"n": 4}));
                    let ms = t0.elapsed().as_secs_f64() * 1000.0;

                    let (ok, fib_ok) = match resp {
                        Ok(r) if r.status().as_u16() == 200 => {
                            let body: serde_json::Value =
                                r.into_body().read_json().unwrap_or_default();
                            (true, body["result"].as_u64() == Some(3))
                        }
                        Ok(r) => {
                            let code = r.status().as_u16();
                            let msg = r.into_body().read_to_string().unwrap_or_default();
                            eprintln!("  [WARN] HTTP {} for {}: {}", code, user, msg);
                            (false, false)
                        }
                        Err(e) => {
                            eprintln!("  [WARN] transport error: {}", e);
                            (false, false)
                        }
                    };

                    rows.lock().unwrap().push(Row {
                        user: user.to_string(),
                        broker: broker.1,
                        latency_ms: ms,
                        ok,
                        fib_ok,
                    });
                }
            }));
        }
        for h in handles {
            h.join().unwrap();
        }
        let wall = clock.elapsed();
        let rows = rows.lock().unwrap();

        // ── Metrics ────────────────────────────────────────────────────
        let n = rows.len();
        let ok = rows.iter().filter(|r| r.ok).count();
        let fib_ok = rows.iter().filter(|r| r.fib_ok).count();
        let rps = n as f64 / wall.as_secs_f64();

        let mut lats: Vec<f64> = rows.iter().map(|r| r.latency_ms).collect();
        lats.sort_by(|a, b| a.partial_cmp(b).unwrap());
        let pct = |p: usize| lats[lats.len() * p / 100];

        let node_stats = |name: &str| {
            let rs: Vec<&Row> = rows.iter().filter(|r| r.broker == name).collect();
            let cnt = rs.len();
            let avg_ms = if cnt > 0 {
                rs.iter().map(|r| r.latency_ms).sum::<f64>() / cnt as f64
            } else {
                0.0
            };
            (cnt, avg_ms)
        };
        let (n1_cnt, n1_lat) = node_stats("Node01");
        let (n2_cnt, n2_lat) = node_stats("Node02");

        // Per-user
        let u_stats: Vec<(&str, usize)> = users
            .iter()
            .map(|u| {
                let cnt = rows.iter().filter(|r| r.user == *u).count();
                (*u, cnt)
            })
            .collect();

        // ── Report ─────────────────────────────────────────────────────
        println!("\n{}", "═".repeat(72));
        println!("  P2P Mesh Fibonacci(4) — 500 Requests (no billing)");
        println!("{}", "═".repeat(72));

        println!("\n  Cluster Topology");
        println!("  {}", "─".repeat(68));
        println!(
            "    Node01  broker=0.0.0.0:{}  own_ip=127.0.0.2  worker=127.0.0.2:{}",
            b1_port, w1_port
        );
        println!(
            "    Node02  broker=0.0.0.0:{}  own_ip=127.0.0.3  worker=127.0.0.3:{}",
            b2_port, w2_port
        );
        println!("    Mode:   broker-to-broker (no API, billing disabled)");

        println!("\n  Execution Summary");
        println!("  {}", "─".repeat(68));
        println!("    Total requests:     {}", n);
        println!(
            "    Successful:         {} ({:.1}%)",
            ok,
            ok as f64 / n as f64 * 100.0
        );
        println!("    fibonacci(4) = 3:   {}/{} correct", fib_ok, n);
        println!("    Wall-clock time:    {:.2}s", wall.as_secs_f64());
        println!("    Throughput:         {:.1} req/s", rps);

        println!("\n  Latency (ms)");
        println!("  {}", "─".repeat(68));
        println!("    Min:    {:.2}", lats[0]);
        println!("    Avg:    {:.2}", lats.iter().sum::<f64>() / n as f64);
        println!("    p50:    {:.2}", pct(50));
        println!("    p90:    {:.2}", pct(90));
        println!("    p99:    {:.2}", pct(99));
        println!("    Max:    {:.2}", lats[n - 1]);

        println!("\n  Per-Node Breakdown");
        println!("  {}", "─".repeat(68));
        println!("    Node01:  {} reqs  avg_lat={:.2}ms", n1_cnt, n1_lat);
        println!("    Node02:  {} reqs  avg_lat={:.2}ms", n2_cnt, n2_lat);

        println!("\n  Per-User Breakdown");
        println!("  {}", "─".repeat(68));
        for (u, cnt) in &u_stats {
            println!("    {:8}  {} reqs", u, cnt);
        }

        // ── API Verification ───────────────────────────────────────────
        println!("\n  {}", "━".repeat(68));
        println!("  API VERIFICATION  (GET /stats, /workers)");
        println!("  {}", "━".repeat(68));

        for (label, base) in [("Node01", &b1_url), ("Node02", &b2_url)] {
            println!("\n  ┌─ {} ({})", label, base);

            // --- /workers ---
            let workers_raw = match agent.get(&format!("{}/workers", base)).call() {
                Ok(r) => r.into_body().read_to_string().unwrap_or_default(),
                Err(e) => format!("ERR: {}", e),
            };
            let workers_resp: serde_json::Value =
                serde_json::from_str(&workers_raw).unwrap_or_default();
            let workers_arr = workers_resp["workers"]
                .as_array()
                .or_else(|| workers_resp.as_array());
            println!(
                "  │  /workers  ({} registered)",
                workers_arr.map(|a| a.len()).unwrap_or(0)
            );
            if let Some(ws) = workers_arr {
                for w in ws {
                    println!(
                        "  │    {} {} status={}",
                        w["name"].as_str().unwrap_or("?"),
                        w["uri"].as_str().unwrap_or("?"),
                        w["status"].as_str().unwrap_or("?")
                    );
                }
            }

            // --- /stats (admin) ---
            let admin_key = "zk_admin_test-master";
            let stats_raw = agent
                .get(&format!("{}/stats", base))
                .header("Authorization", &format!("Bearer {}", admin_key))
                .call()
                .map(|r| r.into_body().read_to_string().unwrap_or_default())
                .unwrap_or_default();
            let stats_resp: serde_json::Value =
                serde_json::from_str(&stats_raw).unwrap_or_default();
            let m = &stats_resp["metrics"];
            println!("  │  /stats");
            println!("  │    total_requests:     {}", m["total_requests"]);
            println!("  │    successful:         {}", m["successful_requests"]);
            println!("  │    failed:             {}", m["failed_requests"]);
            println!(
                "  │    avg_latency_ms:     {:.2}",
                m["avg_latency_ms"].as_f64().unwrap_or(0.0)
            );
            println!("  │    active_workers:     {}", m["active_workers"]);
            println!("  └─");
        }

        println!("\n{}\n", "═".repeat(72));

        // ── Assertions ─────────────────────────────────────────────────
        assert_eq!(ok, n, "all requests must succeed");
        assert_eq!(fib_ok, n, "all fibonacci results must equal 3");
        assert!(rps > 5.0, "throughput must exceed 5 req/s, got {:.1}", rps);
        assert!(n1_cnt > 0, "Node01 must handle requests");
        assert!(n2_cnt > 0, "Node02 must handle requests");

        for (label, base) in [("Node01", &b1_url), ("Node02", &b2_url)] {
            let raw = agent
                .get(&format!("{}/stats", base))
                .header("Authorization", "Bearer zk_admin_test-master")
                .call()
                .map(|r| r.into_body().read_to_string().unwrap_or_default())
                .unwrap_or_default();
            let stats: serde_json::Value = serde_json::from_str(&raw).unwrap_or_default();
            let api_total = stats["metrics"]["total_requests"].as_u64().unwrap_or(0);
            let api_ok = stats["metrics"]["successful_requests"]
                .as_u64()
                .unwrap_or(0);
            assert!(api_total > 0, "{} /stats must report requests", label);
            assert_eq!(
                api_total, api_ok,
                "{} all requests must be successful",
                label
            );
        }
    }

    /// 500 fibonacci(4) requests across a 2-node P2P mesh — through the real API.
    ///
    /// Same topology as test_p2p_mesh_fibonacci_500, but with billing enabled
    /// via ZAKURO_API_URL / ZAKURO_API_KEY.  Credits are enforced and
    /// transactions sync to the centralized dashboard.
    ///
    /// The test resolves the real `zakuro_user_id` from the dashboard so that
    /// flushed transactions reference an existing user and are accepted by the
    /// `POST /api/broker/batch-sync` endpoint.
    #[test]
    fn test_p2p_mesh_fibonacci_500_api() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        use std::time::Instant;

        let api_url = match std::env::var("ZAKURO_API_URL") {
            Ok(v) if !v.is_empty() => v,
            _ => {
                println!("SKIP: ZAKURO_API_URL not set");
                return;
            }
        };
        let api_key = match std::env::var("ZAKURO_API_KEY") {
            Ok(v) if !v.is_empty() => v,
            _ => {
                println!("SKIP: ZAKURO_API_KEY not set");
                return;
            }
        };

        // ── Resolve real zakuro_user_id from the dashboard ──────────────
        let agent = ureq::Agent::new_with_config(
            ureq::Agent::config_builder()
                .timeout_global(Some(Duration::from_secs(10)))
                .http_status_as_error(false)
                .build(),
        );

        let me_raw = match agent
            .get(&format!(
                "{}/api/auth/me/api-key",
                api_url.trim_end_matches('/')
            ))
            .header("Authorization", &format!("Bearer {}", api_key))
            .call()
        {
            Ok(resp) => resp.into_body().read_to_string().unwrap_or_default(),
            Err(e) => {
                println!("SKIP: /api/auth/me/api-key failed ({})", e);
                return;
            }
        };
        let me: serde_json::Value = serde_json::from_str(&me_raw).unwrap_or_default();
        let zakuro_uid = match me["zakuro_user_id"].as_str() {
            Some(uid) => uid,
            None => {
                println!("SKIP: no zakuro_user_id in response");
                return;
            }
        };
        let balance_before = me["credits_balance"].as_f64().unwrap_or(0.0);
        let username = me["username"].as_str().unwrap_or("?");

        println!(
            "  Dashboard user: {} (zakuro_user_id={})",
            username, zakuro_uid
        );
        println!("  Balance before: {:.6} credits", balance_before);

        let peer_key = "fib-mesh-api-key";
        let master = "test-master";
        let total_req = 500usize;
        let threads = 10usize;
        let per_thread = total_req / threads;
        let user_key = format!("zk_{}_0001", zakuro_uid);

        // ── Workers ────────────────────────────────────────────────────
        let (w1_uri, _w1_port) = start_fibonacci_worker("127.0.0.2");
        let (w2_uri, _w2_port) = start_fibonacci_worker("127.0.0.3");

        // ── Brokers (publish-lock model) ──────────────────────────────
        // Startup handshake auto-resolves owner_user_id from API key,
        // so both brokers end up owned by the same verified dashboard user.
        // Each broker registers ONLY its own local worker.
        // Cross-node execution goes through POST /peer/tasks/offer.
        let b1_port = free_port();
        let b2_port = free_port();

        let b1_url = start_mesh_broker(
            b1_port,
            "127.0.0.2",
            "placeholder",
            peer_key,
            vec![format!("http://127.0.0.3:{}", b2_port)],
            Some(api_url.clone()),
            Some(api_key.clone()),
        );
        let b2_url = start_mesh_broker(
            b2_port,
            "127.0.0.3",
            "placeholder",
            peer_key,
            vec![format!("http://127.0.0.2:{}", b1_port)],
            Some(api_url.clone()),
            Some(api_key.clone()),
        );

        // Publish-lock model: each broker registers ONLY its own local worker.
        // Cross-node execution happens when a request arrives at a broker
        // that has no local worker → it publishes the task to its peer.
        //
        // Setup:
        //   zk0node01 (127.0.0.2): has fib-node01 locally
        //   zk0node02 (127.0.0.3): has fib-node02 locally
        //
        // Traffic pattern:
        //   Request → zk0node01 → fib-node01 (local)
        //   Request → zk0node02 → fib-node02 (local)
        //   Request → zk0node01 (no local for fib-node02 → published to zk0node02)
        //   Request → zk0node02 (no local for fib-node01 → published to zk0node01)
        register_worker_at(&b1_url, "fib-node01", &w1_uri, 0.001);
        register_worker_at(&b2_url, "fib-node02", &w2_uri, 0.001);

        // ── Fire 500 requests ──────────────────────────────────────────
        struct Row {
            broker: &'static str,
            worker: String,
            latency_ms: f64,
            cost: f64,
            ok: bool,
            fib_ok: bool,
        }

        let rows = Arc::new(std::sync::Mutex::new(Vec::<Row>::with_capacity(total_req)));
        let idx = Arc::new(AtomicUsize::new(0));
        let clock = Instant::now();

        let mut handles = Vec::new();
        for _ in 0..threads {
            let rows = rows.clone();
            let idx = idx.clone();
            let agent = agent.clone();
            let b1 = b1_url.clone();
            let b2 = b2_url.clone();
            let key = user_key.clone();

            handles.push(thread::spawn(move || {
                for _ in 0..per_thread {
                    let i = idx.fetch_add(1, Ordering::SeqCst);
                    let broker = if i.is_multiple_of(2) {
                        (&b1, "Node01")
                    } else {
                        (&b2, "Node02")
                    };

                    let t0 = Instant::now();
                    let resp = agent
                        .post(&format!("{}/execute", broker.0))
                        .header("Authorization", &format!("Bearer {}", key))
                        .header(
                            "X-Zakuro-Requirements",
                            r#"{"strategy":"round_robin","estimated_duration_secs":0.01}"#,
                        )
                        .send_json(serde_json::json!({"n": 4}));
                    let ms = t0.elapsed().as_secs_f64() * 1000.0;

                    let (ok, cost, fib_ok, worker) = match resp {
                        Ok(r) if r.status().as_u16() == 200 => {
                            let c = r
                                .headers()
                                .get("X-Zakuro-Cost")
                                .and_then(|v| v.to_str().ok())
                                .and_then(|v| v.parse::<f64>().ok())
                                .unwrap_or(0.0);
                            let w = r
                                .headers()
                                .get("X-Zakuro-Worker")
                                .and_then(|v| v.to_str().ok())
                                .unwrap_or("?")
                                .to_string();
                            let body: serde_json::Value =
                                r.into_body().read_json().unwrap_or_default();
                            (true, c, body["result"].as_u64() == Some(3), w)
                        }
                        Ok(r) => {
                            let code = r.status().as_u16();
                            let msg = r.into_body().read_to_string().unwrap_or_default();
                            eprintln!("  [WARN] HTTP {} : {}", code, msg);
                            (false, 0.0, false, "?".to_string())
                        }
                        Err(e) => {
                            eprintln!("  [WARN] transport error: {}", e);
                            (false, 0.0, false, "?".to_string())
                        }
                    };

                    rows.lock().unwrap().push(Row {
                        broker: broker.1,
                        worker,
                        latency_ms: ms,
                        cost,
                        ok,
                        fib_ok,
                    });
                }
            }));
        }
        for h in handles {
            h.join().unwrap();
        }
        let wall = clock.elapsed();
        let rows = rows.lock().unwrap();

        // ── Metrics ────────────────────────────────────────────────────
        let n = rows.len();
        let ok = rows.iter().filter(|r| r.ok).count();
        let fib_ok = rows.iter().filter(|r| r.fib_ok).count();
        let rps = n as f64 / wall.as_secs_f64();
        let charged = rows.iter().filter(|r| r.cost > 0.0).count();
        let _free = rows.iter().filter(|r| r.cost == 0.0 && r.ok).count();
        let tot_cost: f64 = rows.iter().map(|r| r.cost).sum();

        // Cross-node routing: broker received ≠ worker's node
        let n1_to_n2 = rows
            .iter()
            .filter(|r| r.broker == "Node01" && r.worker == "fib-node02")
            .count();
        let n2_to_n1 = rows
            .iter()
            .filter(|r| r.broker == "Node02" && r.worker == "fib-node01")
            .count();
        let n1_local = rows
            .iter()
            .filter(|r| r.broker == "Node01" && r.worker == "fib-node01")
            .count();
        let n2_local = rows
            .iter()
            .filter(|r| r.broker == "Node02" && r.worker == "fib-node02")
            .count();

        let mut lats: Vec<f64> = rows.iter().map(|r| r.latency_ms).collect();
        lats.sort_by(|a, b| a.partial_cmp(b).unwrap());
        let pct = |p: usize| lats[lats.len() * p / 100];

        let node_stats = |name: &str| {
            let rs: Vec<&Row> = rows.iter().filter(|r| r.broker == name).collect();
            let cnt = rs.len();
            let c: f64 = rs.iter().map(|r| r.cost).sum();
            let ch = rs.iter().filter(|r| r.cost > 0.0).count();
            let fr = rs.iter().filter(|r| r.cost == 0.0 && r.ok).count();
            let avg_ms = if cnt > 0 {
                rs.iter().map(|r| r.latency_ms).sum::<f64>() / cnt as f64
            } else {
                0.0
            };
            (cnt, ch, fr, c, avg_ms)
        };
        let (n1_cnt, _n1_ch, _n1_fr, n1_cost, n1_lat) = node_stats("Node01");
        let (n2_cnt, _n2_ch, _n2_fr, n2_cost, n2_lat) = node_stats("Node02");

        // ── Report ─────────────────────────────────────────────────────
        println!("\n{}", "═".repeat(72));
        println!("  P2P Publish-Lock Fibonacci(4) — 500 Requests (API mode)");
        println!("{}", "═".repeat(72));

        println!("\n  Cluster Topology (publish-lock)");
        println!("  {}", "─".repeat(68));
        println!(
            "    zk0node01  broker=0.0.0.0:{}  ip=127.0.0.2  local_worker=fib-node01",
            b1_port
        );
        println!(
            "    zk0node02  broker=0.0.0.0:{}  ip=127.0.0.3  local_worker=fib-node02",
            b2_port
        );
        println!("    Mode:    API (verified owner via dashboard handshake)");
        println!("    Owner:   {} (zakuro_user_id={})", username, zakuro_uid);
        println!("    Model:   publish-lock — each broker executes only on its own workers");

        println!("\n  Execution Routing");
        println!("  {}", "─".repeat(68));
        println!(
            "    zk0node01 → fib-node01 (local):   {} requests",
            n1_local
        );
        println!(
            "    zk0node01 → fib-node02 (peer):    {} requests  (published to zk0node02)",
            n1_to_n2
        );
        println!(
            "    zk0node02 → fib-node02 (local):   {} requests",
            n2_local
        );
        println!(
            "    zk0node02 → fib-node01 (peer):    {} requests  (published to zk0node01)",
            n2_to_n1
        );

        println!("\n  Execution Summary");
        println!("  {}", "─".repeat(68));
        println!("    Total requests:     {}", n);
        println!(
            "    Successful:         {} ({:.1}%)",
            ok,
            ok as f64 / n as f64 * 100.0
        );
        println!("    fibonacci(4) = 3:   {}/{} correct", fib_ok, n);
        println!("    Wall-clock time:    {:.2}s", wall.as_secs_f64());
        println!("    Throughput:         {:.1} req/s", rps);

        println!("\n  Latency (ms)");
        println!("  {}", "─".repeat(68));
        println!("    Min:    {:.2}", lats[0]);
        println!("    Avg:    {:.2}", lats.iter().sum::<f64>() / n as f64);
        println!("    p50:    {:.2}", pct(50));
        println!("    p90:    {:.2}", pct(90));
        println!("    p99:    {:.2}", pct(99));
        println!("    Max:    {:.2}", lats[n - 1]);

        println!("\n  Credits (self-execution = free, verified owner)");
        println!("  {}", "─".repeat(68));
        println!("    All executions free (same verified owner on all nodes)");
        println!("    Total cost:  {:.6} credits", tot_cost);

        println!("\n  Per-Node Breakdown");
        println!("  {}", "─".repeat(68));
        println!(
            "    zk0node01:  {} reqs  cost={:.6}  avg_lat={:.2}ms",
            n1_cnt, n1_cost, n1_lat
        );
        println!(
            "    zk0node02:  {} reqs  cost={:.6}  avg_lat={:.2}ms",
            n2_cnt, n2_cost, n2_lat
        );

        // ── Broker API Verification ────────────────────────────────────
        println!("\n  {}", "━".repeat(68));
        println!("  BROKER API  (GET /stats, /workers, /peer/identity)");
        println!("  {}", "━".repeat(68));

        let admin_key = format!("zk_admin_{}", master);

        for (label, base) in [("zk0node01", &b1_url), ("zk0node02", &b2_url)] {
            println!("\n  ┌─ {} ({})", label, base);

            let workers_raw = match agent.get(&format!("{}/workers", base)).call() {
                Ok(r) => r.into_body().read_to_string().unwrap_or_default(),
                Err(e) => format!("ERR: {}", e),
            };
            let workers_resp: serde_json::Value =
                serde_json::from_str(&workers_raw).unwrap_or_default();
            let workers_arr = workers_resp["workers"]
                .as_array()
                .or_else(|| workers_resp.as_array());
            println!(
                "  │  /workers  ({} registered)",
                workers_arr.map(|a| a.len()).unwrap_or(0)
            );

            let stats_raw = agent
                .get(&format!("{}/stats", base))
                .header("Authorization", &format!("Bearer {}", admin_key))
                .call()
                .map(|r| r.into_body().read_to_string().unwrap_or_default())
                .unwrap_or_default();
            let stats_resp: serde_json::Value =
                serde_json::from_str(&stats_raw).unwrap_or_default();
            let m = &stats_resp["metrics"];
            println!(
                "  │  /stats  total={} ok={} spent={:.6}",
                m["total_requests"],
                m["successful_requests"],
                m["total_credits_spent"].as_f64().unwrap_or(0.0)
            );

            // Verify peer identity (handshake)
            let id_raw = agent
                .get(&format!("{}/peer/identity", base))
                .header("X-Peer-Key", peer_key)
                .call()
                .map(|r| r.into_body().read_to_string().unwrap_or_default())
                .unwrap_or_default();
            let id: serde_json::Value = serde_json::from_str(&id_raw).unwrap_or_default();
            println!(
                "  │  /peer/identity  owner={} verified={} workers={}",
                id["owner_user_id"].as_str().unwrap_or("?"),
                id["verified"].as_bool().unwrap_or(false),
                id["workers"].as_array().map(|a| a.len()).unwrap_or(0)
            );
            println!("  └─");
        }

        // ── Dashboard sync ───────────────────────────────────────────
        println!("\n  {}", "━".repeat(68));
        println!("  DASHBOARD VERIFICATION  ({})", api_url);
        println!("  {}", "━".repeat(68));

        println!("    Waiting for flush cycles (5s)...");
        thread::sleep(Duration::from_secs(5));

        let me_after = agent
            .get(&format!(
                "{}/api/auth/me/api-key",
                api_url.trim_end_matches('/')
            ))
            .header("Authorization", &format!("Bearer {}", api_key))
            .call()
            .expect("failed to call /api/auth/me/api-key after flush")
            .into_body()
            .read_to_string()
            .unwrap();
        let me_after: serde_json::Value = serde_json::from_str(&me_after).unwrap();
        let balance_after = me_after["credits_balance"].as_f64().unwrap_or(0.0);
        let deducted = balance_before - balance_after;

        println!("    Balance before:   {:>12.6} credits", balance_before);
        println!("    Balance after:    {:>12.6} credits", balance_after);
        println!("    Deducted:         {:>12.6} credits", deducted);
        println!(
            "    Owner verified:   {} (handshake confirmed at startup)",
            zakuro_uid
        );

        if (balance_after - balance_before).abs() < 0.0001 {
            println!(
                "    Status:           CORRECT — no credits moved (self-execution, verified owner)"
            );
        } else {
            println!("    Status:           UNEXPECTED — balance changed for self-execution");
        }

        println!("\n{}\n", "═".repeat(72));

        // ── Assertions ─────────────────────────────────────────────────
        assert_eq!(ok, n, "all requests must succeed");
        assert_eq!(fib_ok, n, "all fibonacci results must equal 3");
        assert!(rps > 5.0, "throughput must exceed 5 req/s, got {:.1}", rps);
        assert!(n1_cnt > 0, "zk0node01 must handle requests");
        assert!(n2_cnt > 0, "zk0node02 must handle requests");

        // All executions are free (same verified owner = self-execution)
        assert_eq!(charged, 0, "self-execution must not charge credits");
        assert!(tot_cost == 0.0, "total cost must be 0 for self-execution");

        // Dashboard balance unchanged
        assert!(
            (balance_after - balance_before).abs() < 0.0001,
            "balance must not change for self-execution (before={:.6}, after={:.6})",
            balance_before,
            balance_after
        );
    }

    /// QUIC throughput benchmark — forces cross-node traffic through QUIC.
    ///
    /// Topology:
    ///   Gateway (Node01):  broker only, NO workers  → relays via QUIC
    ///   Executor (Node02): broker + fib-node02 worker
    ///
    /// Phase 1: 500 requests → Gateway → QUIC → Executor (measures QUIC relay)
    /// Phase 2: 500 requests → Executor directly (measures local baseline)
    #[test]
    fn test_quic_throughput_benchmark() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        use std::time::Instant;

        let api_url = match std::env::var("ZAKURO_API_URL") {
            Ok(v) if !v.is_empty() => v,
            _ => {
                println!("SKIP: ZAKURO_API_URL not set");
                return;
            }
        };
        let api_key = match std::env::var("ZAKURO_API_KEY") {
            Ok(v) if !v.is_empty() => v,
            _ => {
                println!("SKIP: ZAKURO_API_KEY not set");
                return;
            }
        };

        let agent = ureq::Agent::new_with_config(
            ureq::Agent::config_builder()
                .timeout_global(Some(Duration::from_secs(30)))
                .http_status_as_error(false)
                .build(),
        );
        let me_raw = match agent
            .get(&format!(
                "{}/api/auth/me/api-key",
                api_url.trim_end_matches('/')
            ))
            .header("Authorization", &format!("Bearer {}", api_key))
            .call()
        {
            Ok(resp) => resp.into_body().read_to_string().unwrap_or_default(),
            Err(e) => {
                println!("SKIP: /api/auth/me/api-key failed ({})", e);
                return;
            }
        };
        let me: serde_json::Value = serde_json::from_str(&me_raw).unwrap_or_default();
        let zakuro_uid = match me["zakuro_user_id"].as_str() {
            Some(uid) => uid,
            None => {
                println!("SKIP: no zakuro_user_id in response");
                return;
            }
        };
        let username = me["username"].as_str().unwrap_or("?");

        let peer_key = "quic-bench-key";
        let total_req = 500usize;
        let threads = 10usize;
        let per_thread = total_req / threads;
        let user_key = format!("zk_{}_0001", zakuro_uid);

        // Worker lives ONLY on Node02's IP
        let (_w2_uri, w2_port) = start_fibonacci_worker("127.0.0.3");
        let w2_uri = format!("http://127.0.0.3:{}", w2_port);

        // Allocate 4 distinct ports: 2 HTTP + 2 QUIC
        let b1_port = free_port();
        let q1_port = free_port();
        let b2_port = free_port();
        let q2_port = free_port();

        // Gateway (Node01): no local workers
        let b1_url = start_mesh_broker_with_quic(
            b1_port,
            "127.0.0.2",
            "placeholder",
            peer_key,
            vec![format!("http://127.0.0.3:{}", b2_port)],
            Some(api_url.clone()),
            Some(api_key.clone()),
            Some(q1_port),
        );
        // Executor (Node02): has the worker
        let b2_url = start_mesh_broker_with_quic(
            b2_port,
            "127.0.0.3",
            "placeholder",
            peer_key,
            vec![format!("http://127.0.0.2:{}", b1_port)],
            Some(api_url.clone()),
            Some(api_key.clone()),
            Some(q2_port),
        );

        // Register worker ONLY on Node02
        register_worker_at(&b2_url, "fib-node02", &w2_uri, 0.001);

        // Wait for QUIC port discovery (brokers query each other's /peer/identity)
        thread::sleep(Duration::from_millis(1500));

        // ── Helper: run N requests against a given broker URL ────────
        #[derive(Debug)]
        struct Row {
            latency_ms: f64,
            ok: bool,
            fib_ok: bool,
            worker: String,
            transport: String,
        }

        let run_phase = |broker_url: &str, n: usize| -> Vec<Row> {
            let rows = Arc::new(std::sync::Mutex::new(Vec::<Row>::with_capacity(n)));
            let idx = Arc::new(AtomicUsize::new(0));
            let mut handles = Vec::new();
            for _ in 0..threads {
                let rows = rows.clone();
                let idx = idx.clone();
                let agent = agent.clone();
                let url = broker_url.to_string();
                let key = user_key.clone();

                handles.push(thread::spawn(move || {
                    for _ in 0..per_thread {
                        let _ = idx.fetch_add(1, Ordering::SeqCst);
                        let t0 = Instant::now();
                        let resp = agent
                            .post(&format!("{}/execute", url))
                            .header("Authorization", &format!("Bearer {}", key))
                            .header(
                                "X-Zakuro-Requirements",
                                r#"{"strategy":"round_robin","estimated_duration_secs":0.01}"#,
                            )
                            .send_json(serde_json::json!({"n": 4}));
                        let ms = t0.elapsed().as_secs_f64() * 1000.0;

                        let (ok, fib_ok, worker, transport) = match resp {
                            Ok(r) if r.status().as_u16() == 200 => {
                                let w = r
                                    .headers()
                                    .get("X-Zakuro-Worker")
                                    .and_then(|v| v.to_str().ok())
                                    .unwrap_or("?")
                                    .to_string();
                                let t = r
                                    .headers()
                                    .get("X-Zakuro-Transport")
                                    .and_then(|v| v.to_str().ok())
                                    .unwrap_or("?")
                                    .to_string();
                                let body: serde_json::Value =
                                    r.into_body().read_json().unwrap_or_default();
                                (true, body["result"].as_u64() == Some(3), w, t)
                            }
                            Ok(r) => {
                                let code = r.status().as_u16();
                                let msg = r.into_body().read_to_string().unwrap_or_default();
                                eprintln!("  [WARN] HTTP {} : {}", code, msg);
                                (false, false, "?".to_string(), "?".to_string())
                            }
                            Err(e) => {
                                eprintln!("  [WARN] transport error: {}", e);
                                (false, false, "?".to_string(), "?".to_string())
                            }
                        };
                        rows.lock().unwrap().push(Row {
                            latency_ms: ms,
                            ok,
                            fib_ok,
                            worker,
                            transport,
                        });
                    }
                }));
            }
            for h in handles {
                h.join().unwrap();
            }
            Arc::try_unwrap(rows).unwrap().into_inner().unwrap()
        };

        // Phase 1: All requests → Gateway (Node01) → must relay via QUIC → Executor (Node02)
        let clock1 = Instant::now();
        let quic_rows = run_phase(&b1_url, total_req);
        let wall1 = clock1.elapsed();

        // Phase 2: All requests → Executor (Node02) → local execution (baseline)
        let clock2 = Instant::now();
        let local_rows = run_phase(&b2_url, total_req);
        let wall2 = clock2.elapsed();

        // ── Metrics helper ───────────────────────────────────────────
        let stats = |rows: &[Row]| {
            let n = rows.len();
            let ok = rows.iter().filter(|r| r.ok).count();
            let fib = rows.iter().filter(|r| r.fib_ok).count();
            let mut lats: Vec<f64> = rows.iter().map(|r| r.latency_ms).collect();
            lats.sort_by(|a, b| a.partial_cmp(b).unwrap());
            let avg = lats.iter().sum::<f64>() / n.max(1) as f64;
            let p50 = lats[n * 50 / 100];
            let p90 = lats[n * 90 / 100];
            let p99 = lats[n * 99 / 100];
            (n, ok, fib, avg, p50, p90, p99, lats[0], lats[n - 1])
        };

        let (qn, qok, qfib, qavg, qp50, qp90, qp99, qmin, qmax) = stats(&quic_rows);
        let (ln, lok, lfib, lavg, lp50, lp90, lp99, lmin, lmax) = stats(&local_rows);
        let qrps = qn as f64 / wall1.as_secs_f64();
        let lrps = ln as f64 / wall2.as_secs_f64();

        let quic_count = quic_rows.iter().filter(|r| r.transport == "quic").count();
        let local_count = local_rows.iter().filter(|r| r.transport == "local").count();

        // ── Report ───────────────────────────────────────────────────
        println!("\n{}", "═".repeat(72));
        println!(
            "  QUIC Throughput Benchmark — {} requests × 2 phases",
            total_req
        );
        println!("{}", "═".repeat(72));

        println!("\n  Cluster Topology");
        println!("  {}", "─".repeat(68));
        println!(
            "    Gateway   (Node01) http=0.0.0.0:{}  quic=0.0.0.0:{}  workers=NONE",
            b1_port, q1_port
        );
        println!(
            "    Executor  (Node02) http=0.0.0.0:{}  quic=0.0.0.0:{}  worker=127.0.0.3:{}",
            b2_port, q2_port, w2_port
        );
        println!(
            "    Owner:     {} (zakuro_user_id={})",
            username, zakuro_uid
        );
        println!("    Protocol:  QUIC (quinn, multiplexed UDP streams)");

        println!(
            "\n  Phase 1: QUIC Relay ({} requests → Gateway → QUIC → Executor)",
            total_req
        );
        println!("  {}", "─".repeat(68));
        println!(
            "    Transport:   {} QUIC / {} other",
            quic_count,
            qn - quic_count
        );
        println!(
            "    Success:     {}/{} ({:.1}%)   fib(4)=3: {}/{}",
            qok,
            qn,
            qok as f64 / qn as f64 * 100.0,
            qfib,
            qn
        );
        println!(
            "    Wall clock:  {:.2}s   Throughput: {:.1} req/s",
            wall1.as_secs_f64(),
            qrps
        );
        println!(
            "    Latency:     min={:.2}  avg={:.2}  p50={:.2}  p90={:.2}  p99={:.2}  max={:.2}",
            qmin, qavg, qp50, qp90, qp99, qmax
        );

        println!(
            "\n  Phase 2: Local Baseline ({} requests → Executor directly)",
            total_req
        );
        println!("  {}", "─".repeat(68));
        println!(
            "    Transport:   {} local / {} other",
            local_count,
            ln - local_count
        );
        println!(
            "    Success:     {}/{} ({:.1}%)   fib(4)=3: {}/{}",
            lok,
            ln,
            lok as f64 / ln as f64 * 100.0,
            lfib,
            ln
        );
        println!(
            "    Wall clock:  {:.2}s   Throughput: {:.1} req/s",
            wall2.as_secs_f64(),
            lrps
        );
        println!(
            "    Latency:     min={:.2}  avg={:.2}  p50={:.2}  p90={:.2}  p99={:.2}  max={:.2}",
            lmin, lavg, lp50, lp90, lp99, lmax
        );

        println!("\n  Comparison");
        println!("  {}", "─".repeat(68));
        let overhead = if lavg > 0.0 {
            (qavg - lavg) / lavg * 100.0
        } else {
            0.0
        };
        println!(
            "    QUIC relay overhead:  {:.2}ms avg ({:+.1}% vs local)",
            qavg - lavg,
            overhead
        );
        println!(
            "    Throughput ratio:     {:.2}x (local/QUIC)",
            lrps / qrps.max(0.01)
        );

        println!("\n{}\n", "═".repeat(72));

        // ── Assertions ───────────────────────────────────────────────
        assert_eq!(qok, qn, "all QUIC-relayed requests must succeed");
        assert_eq!(qfib, qn, "all QUIC fibonacci results must be correct");
        assert_eq!(lok, ln, "all local requests must succeed");
        assert_eq!(lfib, ln, "all local fibonacci results must be correct");
        assert!(
            qrps > 5.0,
            "QUIC throughput must exceed 5 req/s, got {:.1}",
            qrps
        );
        assert!(
            quic_count > 0,
            "at least some requests must use QUIC transport (got {})",
            quic_count
        );
    }

    // ═══════════════════════════════════════════════════════════════════════
    // Billing consistency tests
    // Two-broker WireGuard scenario: validates that all identified billing bugs
    // are fixed and the system works correctly in a distributed setup.
    // ═══════════════════════════════════════════════════════════════════════

    /// Verify that get_balance() populates local_credits so reserve() succeeds.
    ///
    /// Regression for: "Standalone billing always fails — two credit maps never sync"
    /// Before fix: get_balance() stored into authoritative_balances, reserve() read
    /// from local_credits (separate map) → always INSUFFICIENT_CREDITS.
    #[test]
    fn test_standalone_billing_cycle_reserve_commit() {
        use crate::broker::ledger::Ledger;

        let ledger = Ledger::new(None, None);

        // Simulate API fetch: get_balance() now populates local_credits as well
        ledger
            .local_credits
            .insert("user_billing_01".to_string(), 50.0);

        let res = ledger.reserve("user_billing_01", 10.0, "req-standalone-01");
        assert!(
            res.is_ok(),
            "reserve must succeed after balance is in local_credits"
        );

        let balance_mid = ledger
            .local_credits
            .get("user_billing_01")
            .map(|v| *v)
            .unwrap_or(0.0);
        assert_eq!(balance_mid, 40.0, "10 credits held in reservation");

        // Commit: actual cost is 7.5 (reserved 10, refund 2.5)
        let balance_after = ledger.commit(&res.unwrap(), 7.5).unwrap();
        assert!(
            (balance_after - 42.5).abs() < 1e-9,
            "balance_after should be 42.5, got {}",
            balance_after
        );
    }

    /// Verify that cancelling a reservation gives a full refund.
    #[test]
    fn test_standalone_billing_cancel_full_refund() {
        use crate::broker::ledger::Ledger;

        let ledger = Ledger::new(None, None);
        ledger
            .local_credits
            .insert("user_billing_02".to_string(), 30.0);

        let res_id = ledger
            .reserve("user_billing_02", 20.0, "req-cancel-01")
            .unwrap();
        assert_eq!(
            ledger
                .local_credits
                .get("user_billing_02")
                .map(|v| *v)
                .unwrap_or(0.0),
            10.0
        );

        ledger.cancel(&res_id).unwrap();
        assert_eq!(
            ledger
                .local_credits
                .get("user_billing_02")
                .map(|v| *v)
                .unwrap_or(0.0),
            30.0,
            "full refund on cancel"
        );
    }

    /// Verify reserve correctly rejects when balance is insufficient.
    #[test]
    fn test_standalone_billing_insufficient_credits() {
        use crate::broker::ledger::{Ledger, LedgerError};

        let ledger = Ledger::new(None, None);
        ledger.local_credits.insert("user_broke".to_string(), 2.5);

        let result = ledger.reserve("user_broke", 5.0, "req-broke-01");
        match result {
            Err(LedgerError::InsufficientCredits {
                required,
                available,
            }) => {
                assert_eq!(required, 5.0);
                assert_eq!(available, 2.5);
            }
            _ => panic!("expected InsufficientCredits, got {:?}", result.map(|_| ())),
        }
        // Balance must be unchanged
        assert_eq!(
            ledger
                .local_credits
                .get("user_broke")
                .map(|v| *v)
                .unwrap_or(0.0),
            2.5
        );
    }

    /// Verify that reserve() falls back to authoritative_balances when
    /// local_credits is absent (defensive path after the fix).
    #[test]
    fn test_reserve_uses_authoritative_balance_as_fallback() {
        use crate::broker::ledger::Ledger;

        let ledger = Ledger::new(None, None);
        // Populate only the P2P map (authoritative_balances) via local_add_credits
        ledger.local_add_credits("user_authbal", 80.0);
        // local_credits should be empty for this user

        let res = ledger.reserve("user_authbal", 20.0, "req-authbal-01");
        assert!(
            res.is_ok(),
            "reserve should work via authoritative_balances fallback"
        );
    }

    /// Cost calculation must match price_per_hour formula exactly.
    #[test]
    fn test_cost_calculation_price_consistency() {
        use crate::broker::worker::WorkerPricing;

        let pricing = WorkerPricing {
            price_per_hour: 3.6,
            min_charge: 0.001,
        };

        // 10s at 3.6/hr = 3.6/3600*10 = 0.01
        let cost_10s = pricing.estimate_cost(10.0);
        assert!(
            (cost_10s - 0.01).abs() < 1e-9,
            "10s at 3.6/hr should cost exactly 0.01, got {}",
            cost_10s
        );

        // min_charge: 0.1s at 3.6/hr = 0.0001 < 0.001 → floored to 0.001
        let cost_tiny = pricing.estimate_cost(0.1);
        assert_eq!(cost_tiny, pricing.min_charge, "min_charge must apply");

        // Balance accounting: reserve(timeout_cost) → actual < timeout → refund
        let initial = 100.0_f64;
        let reserved = pricing.estimate_cost(300.0); // 5-min reservation
        let actual = pricing.estimate_cost(47.0); // real duration
        let balance_after = initial - actual;
        let simulated = initial - reserved + (reserved - actual);
        assert!(
            (simulated - balance_after).abs() < 1e-10,
            "reserve+refund must equal direct deduction"
        );
    }

    /// P2P authority assignment must be deterministic across repeated calls.
    ///
    /// Regression for: "determine_authority() non-deterministic — DashMap iteration unordered"
    #[test]
    fn test_p2p_authority_is_deterministic() {
        use crate::broker::peer::{Authority, PeerManager};

        let peers = vec!["100.64.0.2:9000".to_string()];
        let pm = PeerManager::new(
            Some("100.64.0.1"),
            &peers,
            9000,
            "test-key".to_string(),
            true,
        );

        // Same user must always resolve to same authority type
        for _ in 0..100 {
            let a1 = pm.determine_authority("9000000001");
            let a2 = pm.determine_authority("9000000001");
            let label = |a: &Authority| match a {
                Authority::Local => "local",
                Authority::Peer(_) => "peer",
                Authority::Standalone => "standalone",
            };
            assert_eq!(
                label(&a1),
                label(&a2),
                "authority changed between calls for same user"
            );
        }
    }

    /// P2P authority URL must be deterministic (same peer URL for same user).
    #[test]
    fn test_p2p_authority_peer_url_stable() {
        use crate::broker::peer::{Authority, PeerManager};

        let peers = vec![
            "100.64.0.3:9000".to_string(),
            "100.64.0.2:9000".to_string(),
            "100.64.0.4:9000".to_string(),
        ];
        let pm1 = PeerManager::new(Some("100.64.0.1"), &peers, 9000, "k".to_string(), true);
        let pm2 = PeerManager::new(Some("100.64.0.1"), &peers, 9000, "k".to_string(), true);

        // Authority must agree across two independently constructed managers
        for uid in &["uid_alpha", "uid_beta", "uid_gamma", "uid_delta"] {
            let a1 = pm1.determine_authority(uid);
            let a2 = pm2.determine_authority(uid);
            // Only the Peer/Peer pair carries a URL to compare. Local and
            // Standalone still have their consistency checked by the type
            // match below.
            if let (Authority::Peer(u1), Authority::Peer(u2)) = (&a1, &a2) {
                assert_eq!(
                    u1, u2,
                    "user {} peer URL must be stable: {} vs {}",
                    uid, u1, u2
                );
            }
            let type1 = match &a1 {
                Authority::Local => 0,
                Authority::Peer(_) => 1,
                Authority::Standalone => 2,
            };
            let type2 = match &a2 {
                Authority::Local => 0,
                Authority::Peer(_) => 1,
                Authority::Standalone => 2,
            };
            assert_eq!(type1, type2, "user {} authority type must be stable", uid);
        }
    }

    /// P2P authority distributes users roughly 50/50 across two brokers.
    #[test]
    fn test_p2p_authority_distribution_two_brokers() {
        use crate::broker::peer::{Authority, PeerManager};

        let pm = PeerManager::new(
            Some("100.64.0.1"),
            &["100.64.0.2:9000".to_string()],
            9000,
            "k".to_string(),
            true,
        );

        let mut local = 0u32;
        let mut remote = 0u32;
        for i in 9000000001u64..9000000101 {
            match pm.determine_authority(&i.to_string()) {
                Authority::Local => local += 1,
                _ => remote += 1,
            }
        }
        assert!(local >= 25, "expected ~50 local, got {}", local);
        assert!(remote >= 25, "expected ~50 remote, got {}", remote);
        assert_eq!(local + remote, 100);
    }

    /// P2P local_reserve/local_commit cycle preserves balance consistency.
    #[test]
    fn test_p2p_local_reserve_commit_balance_consistency() {
        use crate::broker::ledger::Ledger;

        let ledger = Ledger::new(None, None);
        ledger.local_add_credits("user_p2p_01", 200.0);

        // Reserve 50 credits
        let (res_id, bal_before) = ledger
            .local_reserve("user_p2p_01", 50.0, "p2p-req-01")
            .unwrap();
        assert_eq!(bal_before, 200.0);

        // Actual cost 35 → refund 15
        let bal_after = ledger.local_commit(&res_id, 35.0).unwrap();
        assert!(
            (bal_after - 165.0).abs() < 1e-9,
            "balance after commit: expected 165.0, got {}",
            bal_after
        );
        // Total charged = 35, not 50
        assert!(
            ((200.0 - bal_after) - 35.0).abs() < 1e-9,
            "only actual_cost should be debited"
        );
    }

    /// Cross-broker balance sheet: simulate broker_B executing, broker_A authoritative.
    /// Verifies that chaining local_reserve → local_commit gives a consistent ledger.
    #[test]
    fn test_two_broker_balance_sheet_consistency() {
        use crate::broker::ledger::Ledger;
        use crate::broker::worker::WorkerPricing;

        // broker_A is the authority for user_X
        let ledger_a = Ledger::new(None, None);
        ledger_a.local_add_credits("user_X", 100.0);

        // broker_B receives request, calls /peer/reserve on broker_A (simulated here directly)
        let price = WorkerPricing {
            price_per_hour: 3.6,
            min_charge: 0.001,
        };
        let timeout_reservation = price.estimate_cost(300.0); // 5-min max hold

        let (res_id, bal_before) = ledger_a
            .local_reserve("user_X", timeout_reservation, "cross-001")
            .unwrap();
        assert_eq!(bal_before, 100.0);

        // Job finishes in 47.3s → actual cost
        let actual_cost = price.estimate_cost(47.3);
        let bal_after = ledger_a.local_commit(&res_id, actual_cost).unwrap();

        let expected = 100.0 - actual_cost;
        assert!(
            (bal_after - expected).abs() < 1e-9,
            "balance mismatch: expected {:.6}, got {:.6}",
            expected,
            bal_after
        );

        // Dashboard receives only the actual_cost, not the reservation amount
        let total_debited = 100.0 - bal_after;
        assert!(
            (total_debited - actual_cost).abs() < 1e-9,
            "dashboard must see actual_cost ({:.6}), not reservation ({:.6})",
            actual_cost,
            timeout_reservation
        );
    }

    /// Concurrent reservations must not allow negative balances (no overdraft).
    #[test]
    fn test_concurrent_reservations_no_overdraft() {
        use crate::broker::ledger::Ledger;
        use std::sync::Arc;

        let ledger = Arc::new(Ledger::new(None, None));
        ledger
            .local_credits
            .insert("user_concurrent".to_string(), 5.0);

        let handles: Vec<_> = (0..20)
            .map(|i| {
                let l = Arc::clone(&ledger);
                thread::spawn(move || {
                    let _ = l.reserve("user_concurrent", 1.0, &format!("req-c{}", i));
                })
            })
            .collect();
        for h in handles {
            h.join().unwrap();
        }

        let bal = ledger
            .local_credits
            .get("user_concurrent")
            .map(|v| *v)
            .unwrap_or(0.0);
        assert!(bal >= 0.0, "balance went negative: {}", bal);
        assert!(bal <= 5.0, "balance exceeded initial: {}", bal);
    }

    /// Round-robin router distributes requests evenly across two workers.
    #[test]
    fn test_round_robin_even_distribution() {
        use crate::broker::credits::CreditManager;
        use crate::broker::router::{ResourceRequirements, Router, RoutingStrategy};
        use crate::broker::worker::{
            WorkerPricing, WorkerRegistration, WorkerRegistry, WorkerResources,
        };

        let registry = WorkerRegistry::new();
        let credits = CreditManager::new();

        for i in 0..2 {
            let reg = WorkerRegistration {
                name: format!("rr-worker-{}", i),
                uri: format!("http://127.0.0.1:{}", 13960 + i),
                worker_type: "zakuro".to_string(),
                resources: WorkerResources {
                    cpus_available: 4.0,
                    cpus_total: 4.0,
                    memory_available: 8 * 1024 * 1024 * 1024,
                    memory_total: 8 * 1024 * 1024 * 1024,
                    gpus_available: 0,
                    gpus_total: 0,
                },
                pricing: WorkerPricing {
                    price_per_hour: 3.6,
                    min_charge: 0.001,
                },
                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,
            };
            registry.register(reg);
        }
        credits.get_or_create("user_rr", 1000.0);

        let router = Router::new();
        let req = ResourceRequirements {
            strategy: RoutingStrategy::RoundRobin,
            ..Default::default()
        };

        let mut counts: std::collections::HashMap<String, u32> = std::collections::HashMap::new();
        for _ in 0..20 {
            if let Ok(d) = router.select_worker(&registry, &credits, "user_rr", 1000.0, &req) {
                *counts.entry(d.worker.name.clone()).or_insert(0) += 1;
            }
        }

        assert_eq!(counts.len(), 2, "both workers should receive requests");
        for (name, count) in &counts {
            assert!(
                *count >= 5,
                "worker {} only selected {} times in 20",
                name,
                count
            );
        }
    }

    /// WAL recovery: a crash between reserve and commit should cancel the reservation.
    #[test]
    fn test_wal_recovery_cancels_reserved_entries() {
        use crate::broker::ledger::Ledger;
        use crate::broker::recovery;
        use crate::broker::wal::{Wal, WalEntry, WalStatus};

        let tmp = format!("/tmp/zc2-wal-test-{}.jsonl", uuid::Uuid::new_v4());
        let wal = Wal::open(&tmp).expect("WAL open failed");
        let ledger = Ledger::new(None, None);

        // Seed: crash simulated — reserved but never committed
        ledger
            .local_credits
            .insert("user_wal_crash".to_string(), 0.0);

        let entry = WalEntry {
            request_id: "wal-crash-req-001".to_string(),
            user_id: "user_wal_crash".to_string(),
            reservation_id: "wal-crash-res-001".to_string(),
            estimated_cost: 7.5,
            actual_cost: None,
            worker_id: "worker-x".to_string(),
            duration_ms: None,
            timestamp: chrono::Utc::now(),
            status: WalStatus::Reserved,
        };
        wal.append(&entry).expect("append failed");
        wal.flush_buffer().expect("flush failed");

        // Replay: Reserved → cancel → refund
        let tx_buffer = crate::broker::flush::TransactionBuffer::new();
        recovery::replay_wal(&wal, &ledger, &tx_buffer, None);

        let balance = ledger
            .local_credits
            .get("user_wal_crash")
            .map(|v| *v)
            .unwrap_or(0.0);
        assert_eq!(
            balance, 7.5,
            "crashed reservation must be refunded on recovery"
        );

        let _ = std::fs::remove_file(&tmp);
    }

    /// Two-broker HTTP round-robin: verify both brokers receive requests and
    /// credit accounting is consistent across the session.
    ///
    /// This test simulates the WireGuard two-node scenario in-process by
    /// starting two brokers on separate ports, wiring them as peers via HTTP,
    /// and verifying that:
    ///   - All requests succeed (no 402/503)
    ///   - Workers registered on each broker are discoverable
    ///   - Balance decrements match per-worker price×duration
    #[test]
    fn test_two_broker_round_robin_credit_accounting() {
        // Start two brokers on free ports
        let port_a = free_port();
        let port_b = free_port();

        let cfg_a = {
            let mut c = test_config(port_a);
            c.enable_p2p = true;
            c.peer_key = Some("shared-peer-key".to_string());
            c.discovery.peers = vec![format!("127.0.0.1:{}", port_b)];
            c
        };
        let cfg_b = {
            let mut c = test_config(port_b);
            c.enable_p2p = true;
            c.peer_key = Some("shared-peer-key".to_string());
            c.discovery.peers = vec![format!("127.0.0.1:{}", port_a)];
            c
        };

        // Spin up both brokers
        thread::spawn(move || {
            let _ = crate::broker::server::start_server(cfg_a);
        });
        thread::spawn(move || {
            let _ = crate::broker::server::start_server(cfg_b);
        });

        let url_a = format!("http://127.0.0.1:{}", port_a);
        let url_b = format!("http://127.0.0.1:{}", port_b);

        // Wait for both to be healthy
        for (url, port) in [(&url_a, port_a), (&url_b, port_b)] {
            let mut up = false;
            for _ in 0..300 {
                if ureq::get(&format!("{}/health", url))
                    .config()
                    .timeout_global(Some(Duration::from_millis(150)))
                    .build()
                    .call()
                    .is_ok()
                {
                    up = true;
                    break;
                }
                thread::sleep(Duration::from_millis(100));
            }
            assert!(up, "Broker on port {} did not start within 30s", port);
        }

        // Add credits to a test user on broker A (which is the in-memory authority)
        add_credits(&url_a, "user_2broker", 100.0);

        // Register one worker on each broker (but no actual worker process needed —
        // we're just verifying credit accounting and routing decisions, not execution)
        register_worker(&url_a, "node-a-worker", 4.0, 8, 3.6);
        register_worker(&url_b, "node-b-worker", 4.0, 8, 7.2);

        // Wait briefly for peer discovery
        thread::sleep(Duration::from_millis(500));

        // Verify peer health endpoint works between brokers
        let health_a = ureq::get(&format!("{}/peer/health", url_a))
            .header("X-Peer-Key", "shared-peer-key")
            .config()
            .timeout_global(Some(Duration::from_secs(3)))
            .build()
            .call();
        assert!(
            health_a
                .map(|r| r.status().as_u16() == 200)
                .unwrap_or(false),
            "broker_a /peer/health must respond 200"
        );

        let health_b = ureq::get(&format!("{}/peer/health", url_b))
            .header("X-Peer-Key", "shared-peer-key")
            .config()
            .timeout_global(Some(Duration::from_secs(3)))
            .build()
            .call();
        assert!(
            health_b
                .map(|r| r.status().as_u16() == 200)
                .unwrap_or(false),
            "broker_b /peer/health must respond 200"
        );

        // Verify /peer/workers returns workers from each broker
        let workers_a_via_peer = ureq::get(&format!("{}/peer/workers", url_a))
            .header("X-Peer-Key", "shared-peer-key")
            .config()
            .timeout_global(Some(Duration::from_secs(3)))
            .build()
            .call()
            .unwrap()
            .into_body()
            .read_json::<serde_json::Value>()
            .unwrap();
        assert!(
            workers_a_via_peer["workers"]
                .as_array()
                .map(|a| !a.is_empty())
                .unwrap_or(false),
            "broker_a must expose its workers via /peer/workers"
        );

        let workers_b_via_peer = ureq::get(&format!("{}/peer/workers", url_b))
            .header("X-Peer-Key", "shared-peer-key")
            .config()
            .timeout_global(Some(Duration::from_secs(3)))
            .build()
            .call()
            .unwrap()
            .into_body()
            .read_json::<serde_json::Value>()
            .unwrap();
        assert!(
            workers_b_via_peer["workers"]
                .as_array()
                .map(|a| !a.is_empty())
                .unwrap_or(false),
            "broker_b must expose its workers via /peer/workers"
        );
    }

    // ── zc agent (spec 2026-09-13-macos-widget §9) ───────────────────────
    //
    // Real processes: the test binary re-executes itself as the broker and as
    // fake workers (see src/agent/testbin.rs). Timings are shrunk so the whole
    // group finishes in well under a minute. Unix-only, like `crate::agent`.
    #[cfg(unix)]
    mod agent_it {
        use super::{free_port, port_range};
        use crate::agent::config::AgentConfig;
        use crate::agent::core::tests::{hub_json, row};
        use crate::agent::files::tests::tmp;
        use crate::agent::hub::tests::mock_hub;
        use crate::agent::run::start_background;
        use crate::agent::testbin::self_exec_argv;
        use std::path::{Path, PathBuf};
        use std::sync::{Arc, Mutex};
        use std::time::{Duration, Instant};

        /// Every test agent in this module is configured for this many worker
        /// slots (`cfg.max_workers`); `port_block` reserves enough consecutive
        /// ports for exactly that many workers, plus headroom for the
        /// broker's discovery scan.
        const TEST_MAX_WORKERS: u32 = 3;

        /// `TEST_MAX_WORKERS + 20` consecutive free loopback ports, drawn from
        /// the same non-ephemeral counter `free_port()` uses (never from the
        /// OS ephemeral range, which risks a TIME_WAIT collision): worker i
        /// listens on base+i, and the broker scans base..=base+max+19.
        fn port_block() -> u16 {
            port_range(TEST_MAX_WORKERS as u16 + 20)
        }

        fn agent_cfg(workers: u32, sharing: bool) -> (AgentConfig, PathBuf) {
            let state = tmp("agent-it");
            std::fs::create_dir_all(state.join("agent")).unwrap();
            std::fs::write(
                state.join("agent").join("state.json"),
                format!(r#"{{"sharing":{sharing},"workers":{workers},"known_worker_ids":[]}}"#),
            )
            .unwrap();
            let mut cfg = AgentConfig::for_dirs(state.clone());
            cfg.port = 0;
            cfg.broker_port = free_port();
            cfg.base_worker_port = port_block();
            cfg.max_workers = TEST_MAX_WORKERS;
            cfg.broker_argv = self_exec_argv("broker_main");
            cfg.worker_template = self_exec_argv("fake_worker_main");
            cfg.worker_template_overridden = true;
            cfg.child_env = vec![
                ("ZAKURO_HOME".into(), state.display().to_string()),
                (
                    "ZAKURO_WAL_PATH".into(),
                    state.join("wal.jsonl").display().to_string(),
                ),
                ("ZAKURO_SCAN_INTERVAL".into(), "1".into()),
                ("ZAKURO_P2P".into(), "false".into()),
            ];
            cfg.timings.reconcile = Duration::from_millis(300);
            cfg.timings.hub_poll = Duration::from_millis(500);
            cfg.timings.kill_grace = Duration::from_secs(2);
            cfg.timings.broker_stop_wait = Duration::from_secs(5);
            cfg.timings.shutdown_wait = Duration::from_secs(3);
            (cfg, state)
        }

        fn sign_in(state: &Path, hub: &str) {
            std::fs::write(
                state.join("credentials"),
                format!("api_key=zk_1_test\napi_url={hub}\n"),
            )
            .unwrap();
        }

        /// A hub that accepts a connection but never answers it, so whatever
        /// task is waiting on it would block forever unless something aborts
        /// it outright instead of joining it.
        fn hanging_hub() -> String {
            let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap();
            let port = listener.local_addr().unwrap().port();
            std::thread::spawn(move || {
                for stream in listener.incoming().flatten() {
                    std::thread::sleep(Duration::from_secs(120));
                    drop(stream);
                }
            });
            format!("http://127.0.0.1:{port}")
        }

        struct Agent {
            port: u16,
            token: String,
            stop: Option<std::sync::mpsc::Sender<()>>,
            thread: Option<std::thread::JoinHandle<()>>,
        }

        /// Signals the background agent to shut down and waits for it to
        /// finish, so its children never overlap the next test.
        impl Drop for Agent {
            fn drop(&mut self) {
                drop(self.stop.take());
                if let Some(t) = self.thread.take() {
                    let _ = t.join();
                }
            }
        }

        fn start(cfg: AgentConfig) -> Agent {
            let (port, token, stop, thread) = start_background(cfg).expect("agent starts");
            Agent {
                port,
                token,
                stop: Some(stop),
                thread: Some(thread),
            }
        }

        fn http() -> ureq::Agent {
            ureq::Agent::new_with_config(
                ureq::Agent::config_builder()
                    .http_status_as_error(false)
                    .timeout_global(Some(Duration::from_secs(10)))
                    .build(),
            )
        }

        fn summary(a: &Agent) -> serde_json::Value {
            http()
                .get(&format!("http://127.0.0.1:{}/v1/summary", a.port))
                .header("Authorization", format!("Bearer {}", a.token).as_str())
                .call()
                .unwrap()
                .into_body()
                .read_json()
                .unwrap()
        }

        fn send(
            a: &Agent,
            method: &str,
            path: &str,
            body: serde_json::Value,
        ) -> (u16, serde_json::Value) {
            let url = format!("http://127.0.0.1:{}{path}", a.port);
            let auth = format!("Bearer {}", a.token);
            let req = if method == "PUT" {
                http().put(&url)
            } else {
                http().post(&url)
            };
            let resp = req
                .header("Authorization", auth.as_str())
                .send_json(body)
                .unwrap();
            let status = resp.status().as_u16();
            (status, resp.into_body().read_json().unwrap_or_default())
        }

        fn wait_until(what: &str, timeout: Duration, mut done: impl FnMut() -> bool) {
            let end = Instant::now() + timeout;
            while Instant::now() < end {
                if done() {
                    return;
                }
                std::thread::sleep(Duration::from_millis(200));
            }
            panic!("timed out waiting for: {what}");
        }

        fn broker_workers(broker_port: u16) -> Vec<serde_json::Value> {
            let body: serde_json::Value = match http()
                .get(&format!("http://127.0.0.1:{broker_port}/workers"))
                .call()
            {
                Ok(r) => r.into_body().read_json().unwrap_or_default(),
                Err(_) => return vec![],
            };
            body["workers"].as_array().cloned().unwrap_or_default()
        }

        fn broker_worker_names(broker_port: u16) -> Vec<String> {
            let mut names: Vec<String> = broker_workers(broker_port)
                .iter()
                .filter_map(|w| w["name"].as_str().map(String::from))
                .collect();
            names.sort();
            names
        }

        fn worker_status(broker_port: u16, name: &str) -> Option<String> {
            broker_workers(broker_port)
                .into_iter()
                .find(|w| w["name"] == name)
                .and_then(|w| w["status"].as_str().map(String::from))
        }

        fn worker_id(broker_port: u16, name: &str) -> Option<String> {
            broker_workers(broker_port)
                .into_iter()
                .find(|w| w["name"] == name)
                .and_then(|w| w["id"].as_str().map(String::from))
        }

        fn fake_worker_count(worker_port: u16, endpoint: &str) -> u64 {
            http()
                .get(&format!("http://127.0.0.1:{worker_port}/{endpoint}"))
                .call()
                .unwrap()
                .into_body()
                .read_json::<serde_json::Value>()
                .unwrap()["count"]
                .as_u64()
                .unwrap()
        }

        fn fake_worker_execute_count(worker_port: u16) -> u64 {
            fake_worker_count(worker_port, "execute_count")
        }

        fn fake_worker_info_count(worker_port: u16) -> u64 {
            fake_worker_count(worker_port, "info_count")
        }

        /// Submit a keyless local-mode job straight to the test broker's
        /// `/execute` (server.rs:991), the same path real dispatch uses; the
        /// broker load-balances it to whichever registered worker is healthy.
        fn broker_execute(broker_port: u16) {
            let resp = http()
                .post(&format!("http://127.0.0.1:{broker_port}/execute"))
                .send_json(serde_json::json!({"fn": "noop", "args": []}))
                .unwrap();
            assert_eq!(
                resp.status().as_u16(),
                200,
                "the broker must accept a keyless local-mode /execute"
            );
        }

        #[test]
        fn agent_serves_a_summary_merged_from_the_hub_and_its_broker() {
            let (mut cfg, state) = agent_cfg(1, true);
            let pk = cfg.node_pubkey();
            let (hub, _) = mock_hub(move |_, url, _| {
                if url.starts_with("/api/provide/summary") {
                    (200, hub_json(&pk, &row(11, Some(18.0))))
                } else {
                    (404, "{}".into())
                }
            });
            sign_in(&state, &hub);
            cfg.api_url = hub.clone();
            let a = start(cfg);

            wait_until(
                "the broker and one worker running",
                Duration::from_secs(40),
                || {
                    let s = summary(&a);
                    s["this_mac"]["broker"] == "running" && s["this_mac"]["workers"]["running"] == 1
                },
            );
            let s = summary(&a);
            assert_eq!(s["v"], 1);
            assert_eq!(s["env"], "custom");
            assert_eq!(s["account"]["username"], "jean");
            assert_eq!(s["devices"][0]["this_mac"], true);
            assert_eq!(s["prices"]["this_mac"]["state"], "set");
            assert_eq!(s["links"]["hub"], "https://stg.hub.zakuro-ai.com/provide");
            let children: crate::agent::files::Children =
                crate::agent::files::load_json(&state.join("agent").join("children.json")).unwrap();
            assert!(children.broker.is_some() && children.workers.contains_key("0"));
        }

        #[test]
        fn scaling_two_to_one_drains_then_stops_the_highest_worker_and_pause_stops_all() {
            let (mut cfg, state) = agent_cfg(2, true);
            let pk = cfg.node_pubkey();
            let (hub, _) = mock_hub(move |_, _, _| (200, hub_json(&pk, "")));
            sign_in(&state, &hub);
            cfg.api_url = hub;
            let (broker_port, fp8) = (cfg.broker_port, cfg.node_fp8());
            let (w0, w1) = (format!("{fp8}-w0"), format!("{fp8}-w1"));
            let a = start(cfg);

            wait_until(
                "the broker lists w0 and w1",
                Duration::from_secs(40),
                || broker_worker_names(broker_port) == vec![w0.clone(), w1.clone()],
            );
            let (status, s) = send(&a, "PUT", "/v1/workers", serde_json::json!({"count": 1}));
            assert_eq!(
                (status, s["this_mac"]["workers"]["desired"].as_u64()),
                (200, Some(1))
            );
            wait_until(
                "w1 drained, stopped and deleted from the broker",
                Duration::from_secs(30),
                || {
                    broker_worker_names(broker_port) == vec![w0.clone()]
                        && summary(&a)["this_mac"]["workers"]["running"] == 1
                },
            );

            let (status, _) = send(&a, "PUT", "/v1/sharing", serde_json::json!({"on": false}));
            assert_eq!(status, 200);
            wait_until("everything stopped", Duration::from_secs(30), || {
                let s = summary(&a);
                s["this_mac"]["broker"] == "stopped" && s["this_mac"]["workers"]["running"] == 0
            });
        }

        /// spec §9: draining a worker directly on the broker must survive
        /// discovery's periodic re-scan — discovery must never lift a
        /// draining worker back to healthy — and the drained worker must
        /// receive no further `/execute` calls.
        ///
        /// The positive control needs the test broker's keyless local-mode
        /// `/execute`, which the broker only offers off the WireGuard mesh
        /// (`crate::broker::discovery::detect_discovery_mode`): on a mesh
        /// host this is skipped rather than failed, using the same
        /// mesh-presence check the broker itself uses to pick its mode.
        #[test]
        fn draining_a_worker_survives_discovery_rescans_and_stops_new_execute_calls() {
            // The same call the broker itself makes to decide its mode
            // (server.rs's start_server, right before it sets local_mode).
            // Only the positive-control half below needs it: a mesh host
            // still gets the "still draining after discovery rescans" half.
            let local_mode = crate::broker::discovery::detect_discovery_mode(
                &crate::broker::discovery::DiscoveryConfig::default().subnet,
            ) == crate::broker::discovery::DiscoveryMode::Local;

            let (mut cfg, state) = agent_cfg(2, true);
            let pk = cfg.node_pubkey();
            let (hub, _) = mock_hub(move |_, _, _| (200, hub_json(&pk, "")));
            sign_in(&state, &hub);
            cfg.api_url = hub;
            let (broker_port, base_worker_port, fp8) =
                (cfg.broker_port, cfg.base_worker_port, cfg.node_fp8());
            let (w0, w1) = (format!("{fp8}-w0"), format!("{fp8}-w1"));
            let (w0_port, w1_port) = (base_worker_port, base_worker_port + 1);
            let _a = start(cfg); // keeps the agent (and its broker/workers) alive for the scope of this test

            wait_until(
                "the broker lists w0 and w1",
                Duration::from_secs(40),
                || broker_worker_names(broker_port) == vec![w0.clone(), w1.clone()],
            );

            let id1 = worker_id(broker_port, &w1).expect("w1 registered");
            let resp = http()
                .post(&format!(
                    "http://127.0.0.1:{broker_port}/workers/{id1}/drain"
                ))
                .send_empty()
                .unwrap();
            assert_eq!(resp.status().as_u16(), 200);
            assert_eq!(worker_status(broker_port, &w1).as_deref(), Some("draining"));

            let execute_at_drain = fake_worker_execute_count(w1_port);
            let info_at_drain = fake_worker_info_count(w1_port);

            // ZAKURO_SCAN_INTERVAL=1 in cfg.child_env: an observed count of at
            // least 3 more /info probes on w1 proves 3+ discovery re-scans
            // actually happened, rather than assuming it from a fixed sleep.
            wait_until(
                "at least 3 more discovery scans reach w1",
                Duration::from_secs(15),
                || fake_worker_info_count(w1_port) >= info_at_drain + 3,
            );

            assert_eq!(
                worker_status(broker_port, &w1).as_deref(),
                Some("draining"),
                "discovery must not lift a draining worker back to healthy"
            );

            if !local_mode {
                eprintln!(
                    "skipping the /execute positive control in \
                     draining_a_worker_survives_discovery_rescans_and_stops_new_execute_calls: \
                     this host is on the WireGuard mesh, so the test broker's /execute is not \
                     keyless local-mode and would 401"
                );
                return;
            }

            // Positive control: dispatch real jobs through the broker's
            // keyless local-mode /execute (the same path real traffic uses)
            // and prove they land on w0, never on the drained w1.
            let w0_before = fake_worker_execute_count(w0_port);
            const N: u64 = 3;
            for _ in 0..N {
                broker_execute(broker_port);
            }
            assert_eq!(
                fake_worker_execute_count(w0_port),
                w0_before + N,
                "dispatch must still reach the healthy worker"
            );
            assert_eq!(
                fake_worker_execute_count(w1_port),
                execute_at_drain,
                "a draining worker must receive no /execute calls"
            );
        }

        /// Shutdown must SIGTERM every child group and complete within
        /// `shutdown_wait`, plus what a reconcile tick already in flight may
        /// still spend on its two broker checks, plus a small margin, no
        /// matter what else is going on: a hub that never answers must be
        /// abandoned, not waited on, and a worker that ignores TERM must
        /// still end up SIGKILLed.
        #[test]
        fn shutdown_signals_a_term_ignoring_worker_promptly_even_with_a_hanging_hub() {
            let (mut cfg, state) = agent_cfg(1, true);
            let hub = hanging_hub();
            sign_in(&state, &hub);
            cfg.api_url = hub;
            cfg.worker_template = vec![
                "sh".into(),
                "-c".into(),
                "trap '' TERM; exec sleep 30".into(),
            ];
            cfg.worker_template_overridden = true;
            // Much larger than shutdown_wait: if the hub task were joined
            // instead of aborted, shutdown would take this long.
            cfg.timings.hub_timeout = Duration::from_secs(20);
            cfg.timings.hub_poll = Duration::from_millis(300);
            cfg.timings.shutdown_wait = Duration::from_secs(2);
            let shutdown_wait = cfg.timings.shutdown_wait;

            let a = start(cfg);
            wait_until("the worker is running", Duration::from_secs(15), || {
                summary(&a)["this_mac"]["workers"]["running"] == 1
            });
            let children: crate::agent::files::Children =
                crate::agent::files::load_json(&state.join("agent").join("children.json")).unwrap();
            let pid = children.workers.get("0").expect("worker 0 recorded").pid;
            assert!(
                crate::agent::supervisor::is_alive(pid),
                "the worker must be running before shutdown"
            );

            let began = Instant::now();
            // Agent's Drop signals shutdown and joins the background thread,
            // so this line blocks until the whole shutdown has finished.
            drop(a);
            let elapsed = began.elapsed();

            // A tick in flight when shutdown starts only looks at the stop
            // flag once its `/health` and `GET /workers` checks return; on a
            // loaded CI runner each can take its whole timeout.
            let in_flight_tick = 2 * crate::agent::supervisor::BROKER_CHECK_TIMEOUT;
            assert!(
                elapsed < shutdown_wait + in_flight_tick + Duration::from_secs(3),
                "shutdown took {elapsed:?}, expected to finish within shutdown_wait \
                 ({shutdown_wait:?}) plus an in-flight tick's broker checks \
                 ({in_flight_tick:?}) plus a small margin, even with a hanging hub \
                 and a TERM-ignoring worker"
            );
            assert!(
                !crate::agent::supervisor::is_alive(pid),
                "the TERM-ignoring worker group must be gone after shutdown"
            );
            // No reconcile tick may run — let alone respawn anything — once
            // shutdown has started: `children.json`, rewritten last by
            // `shutdown_children`, must show nothing left.
            let children_after: crate::agent::files::Children =
                crate::agent::files::load_json(&state.join("agent").join("children.json")).unwrap();
            assert!(
                children_after.broker.is_none() && children_after.workers.is_empty(),
                "no reconcile tick may respawn children after shutdown: {children_after:?}"
            );
        }

        #[test]
        fn device_price_fans_out_over_http_and_seeds_a_new_row() {
            let (mut cfg, state) = agent_cfg(1, false); // nothing to spawn
            let pk = cfg.node_pubkey();
            let rows = Arc::new(Mutex::new(format!("{},{}", row(11, None), row(12, None))));
            let r = rows.clone();
            let (hub, seen) = mock_hub(move |method, url, _| match method {
                "GET" if url.starts_with("/api/provide/summary") => {
                    (200, hub_json(&pk, &r.lock().unwrap()))
                }
                "PUT" => (200, "{}".into()),
                _ => (404, "{}".into()),
            });
            sign_in(&state, &hub);
            cfg.api_url = hub;
            let a = start(cfg);
            wait_until("the first hub poll", Duration::from_secs(10), || {
                summary(&a)["account"]["username"] == "jean"
            });

            let (status, _) = send(
                &a,
                "PUT",
                "/v1/price",
                serde_json::json!({"scope": "device", "price_per_hour": 18}),
            );
            assert_eq!(status, 200);
            let put_urls = || -> Vec<String> {
                seen.lock()
                    .unwrap()
                    .iter()
                    .filter(|e| e.0 == "PUT")
                    .map(|e| e.1.clone())
                    .collect()
            };
            // The fan-out is concurrent, so the rows arrive in either order.
            let mut fanned_out = put_urls();
            fanned_out.sort();
            assert_eq!(
                fanned_out,
                vec!["/api/workers/11/price", "/api/workers/12/price"]
            );

            *rows.lock().unwrap() = format!(
                "{},{},{}",
                row(11, Some(18.0)),
                row(12, Some(18.0)),
                row(13, None)
            );
            let (status, _) = send(&a, "POST", "/v1/refresh", serde_json::json!({}));
            assert_eq!(status, 200);
            wait_until("row 13 seeded at 18", Duration::from_secs(10), || {
                put_urls().last().map(String::as_str) == Some("/api/workers/13/price")
            });
        }
    }
}