//! `MultiProcCluster`: the true one-process-per-region tier-3 harness (m8p10).
//!
//! # What this harness fixes
//!
//! `cluster_e2e.rs` spawned `n` `tidal-server cluster` processes, but EVERY
//! process held ALL regions (a single-process `ClusterState` per process), so a
//! write to `nodes[0]` replicated inside node 0's own fabric and never crossed
//! into `nodes[1]`'s process. That harness's own module docs flag this as the
//! gap. `MultiProcCluster` closes it: each process runs `tidal-server cluster
//! --region <name>` and owns exactly ONE region (a `ShardReplica`), peering
//! with siblings over real gRPC and forwarding over real HTTP. Convergence is
//! therefore verified against EVERY follower process's own `/cluster/status/local`,
//! not a single node's internal view.
//!
//! # Topology model
//!
//! Every region declares both a `grpc_addr` and an `http_addr` (multi-process
//! mode requires both). The harness allocates a free loopback port for each, then
//! writes a per-process topology file:
//!
//! * a process's OWN region entry always lists its REAL bind addresses (so the
//!   process binds the port the harness opened);
//! * every PEER region entry lists the PUBLISHED address from the
//!   [`AddrRewrite`] hook (identity by default; task 05 interposes TCP proxies by
//!   returning a proxy address for a peer's real address).
//!
//! With the identity hook every process's topology file is byte-identical and
//! uses the real ports — a plain shared topology. The per-process files are how
//! the rewrite hook stays surgical: a proxy only sits on the path peers use to
//! reach a region, never on the region's own self-bind.
//!
//! # Budgets
//!
//! Tier-3 over real OS processes: 60s boot (each process opens a `TidalDb`, builds
//! a gRPC transport on its own runtime, and the suite may spawn several at once),
//! 30s convergence (WAL relay over loopback gRPC). Every wait polls with a
//! deadline; there are no bare sleeps used as a correctness gate (a small bounded
//! settle sleep after spawn is the only sleep, and it never gates correctness).

use std::{
    collections::HashMap,
    fmt::Write as _,
    io::Write as _,
    net::{SocketAddr, TcpListener},
    path::{Path, PathBuf},
    process::{Child, Command},
    sync::{Mutex, OnceLock},
    time::{Duration, Instant},
};

/// Tier-3 boot budget: every process opens a `TidalDb` and builds a gRPC
/// transport on its own runtime, and the suite may spawn several concurrently.
/// Default 60s; override with `TIDAL_TEST_BOOT_BUDGET_SECS` on a slow runner.
pub fn boot_budget() -> Duration {
    env_budget("TIDAL_TEST_BOOT_BUDGET_SECS", 60)
}

/// Tier-3 convergence budget for the WAL relay over loopback gRPC.
/// Default 30s; override with `TIDAL_TEST_CONVERGENCE_BUDGET_SECS` on a slow
/// runner.
pub fn convergence_budget() -> Duration {
    env_budget("TIDAL_TEST_CONVERGENCE_BUDGET_SECS", 30)
}

/// Resolve a budget override from the environment (whole seconds, > 0), or the
/// compiled-in default.
fn env_budget(var: &str, default_secs: u64) -> Duration {
    let secs = std::env::var(var)
        .ok()
        .and_then(|v| v.trim().parse::<u64>().ok())
        .filter(|&s| s > 0)
        .unwrap_or(default_secs);
    Duration::from_secs(secs)
}

/// How long a tripped gRPC circuit breaker stays open before allowing a probe —
/// mirrors `GrpcTransportConfig::default().circuit_breaker_reset` (30s), which
/// is what the spawned `tidal-server` processes run with. Suites that wait for
/// breaker-mediated recovery budget this much on top of [`convergence_budget`].
pub const BREAKER_RESET: Duration = Duration::from_secs(30);

/// Poll cadence for every deadline-bounded wait loop in the harness.
const POLL_INTERVAL: Duration = Duration::from_millis(100);

/// Bounded settle pause after spawning all processes, before the first health
/// poll. Never a correctness gate (health/convergence are polled with deadlines);
/// it just avoids hammering a socket that has not started listening yet.
const SETTLE: Duration = Duration::from_millis(150);

/// Which address a topology entry carries, for the [`AddrRewrite`] hook.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum AddrKind {
    /// The gRPC replication address (`grpc_addr`).
    Grpc,
    /// The public HTTP address (`http_addr`).
    Http,
}

/// Maps a peer region's REAL bind address to the address the OBSERVER process
/// should publish for it.
///
/// `(observer_region, peer_region, kind, real_addr) -> published_addr`. The default
/// is identity (`real_addr.to_string()`). Task 05 supplies a rewrite that returns a
/// TCP-proxy address for a peer's real address so severing the proxy is a real
/// network partition while the process keeps binding its real port.
///
/// The `observer_region` is the region whose topology file is being written (the
/// process that will DIAL the published address). Carrying it lets a rewrite
/// interpose a proxy on a SPECIFIC directed edge (`observer → peer`) — the
/// granularity the follower↔follower partition test needs (cut eu-west↔ap-south
/// while leaving leader↔ap-south intact). A rewrite that only cares about the
/// target ignores `observer_region`.
pub type AddrRewrite = Box<dyn Fn(&str, &str, AddrKind, SocketAddr) -> String + Send + Sync>;

/// Identity rewrite: peers publish the real address verbatim.
#[must_use]
pub fn identity_rewrite() -> AddrRewrite {
    Box::new(|_observer, _peer, _kind, addr| addr.to_string())
}

/// DNS-hostname rewrite (m11p5 §1, exit gate 4): every PEER `grpc_addr` is
/// published as `localhost:<port>` — a HOSTNAME, not a literal `SocketAddr` — so
/// the full dial stack must route through the DNS resolver (`Channel::from_shared`
/// re-resolution, the entire point of the `String` peer retype). The loopback
/// loses no reachability (`localhost` resolves to `127.0.0.1`), but a peer entry
/// that the pre-m11p5 `SocketAddr::parse` would have REJECTED at boot now must
/// boot, replicate, and survive restarts — proving the resolver is in the path.
///
/// HTTP peer entries keep the real IP: the test client dials each node's HTTP
/// addr DIRECTLY (no rewrite on the operator console), and forwarding between
/// nodes is exercised separately; the gate is the gRPC *replication* dial. A
/// process's OWN region entry is never rewritten ([`write_topology_for`]), so the
/// bind side stays a real `0.0.0.0`/loopback `SocketAddr` and only the advertised
/// peer name is a hostname — the exact bind/advertise split §1 specifies.
#[must_use]
pub fn hostname_rewrite() -> AddrRewrite {
    Box::new(|_observer, _peer, kind, addr| match kind {
        AddrKind::Grpc => format!("localhost:{}", addr.port()),
        AddrKind::Http => addr.to_string(),
    })
}

/// Options for [`MultiProcCluster::start_with`].
pub struct ClusterOptions {
    /// Number of regions (one OS process each). Must be >= 2.
    pub regions: usize,
    /// Per-node extra environment, keyed by region INDEX (0-based, topology
    /// order). Merged onto the base env at spawn (e.g. `TIDAL_HLC_SKEW_MS`,
    /// a version tag) — used by the clock-skew / rolling-upgrade scenarios.
    pub extra_env: HashMap<usize, Vec<(String, String)>>,
    /// Peer-address rewrite hook (default identity). See [`AddrRewrite`].
    pub rewrite: AddrRewrite,
    /// Per-node `TIDAL_SERVER_LOG` value (default `"warn"`).
    pub log: String,
    /// Extra topology YAML appended verbatim to every per-process topology
    /// file (default none). Used to tune the m11p4 `election:` block — e.g.
    /// faster timeouts for the failover gates, or `auto_election: false` for
    /// suites that exercise the legacy manual-promote protocol.
    pub topology_extra: Option<String>,
}

impl ClusterOptions {
    /// Default options for `n` regions: no extra env, identity rewrite, `warn` logs.
    #[must_use]
    pub fn new(regions: usize) -> Self {
        Self {
            regions,
            extra_env: HashMap::new(),
            rewrite: identity_rewrite(),
            log: "warn".into(),
            topology_extra: None,
        }
    }

    /// Append extra YAML (e.g. an `election:` block) to every topology file.
    /// Builder-style; chainable.
    #[must_use]
    pub fn with_topology_extra(mut self, yaml: &str) -> Self {
        self.topology_extra = Some(yaml.to_string());
        self
    }

    /// Add extra env for one region (by index). Builder-style; chainable.
    #[must_use]
    pub fn with_env(mut self, region_idx: usize, key: &str, value: &str) -> Self {
        self.extra_env
            .entry(region_idx)
            .or_default()
            .push((key.to_string(), value.to_string()));
        self
    }

    /// Replace the peer-address rewrite hook. Builder-style; chainable.
    #[must_use]
    pub fn with_rewrite(mut self, rewrite: AddrRewrite) -> Self {
        self.rewrite = rewrite;
        self
    }
}

/// One region's static identity within the cluster: its name, real bind ports,
/// and the data dir its `TidalDb` persists to (so a restart reuses the same one).
#[derive(Clone)]
struct RegionPlan {
    name: String,
    grpc: SocketAddr,
    http: SocketAddr,
    data_dir: PathBuf,
    /// Per-shard gRPC bind ports for the m11p6 sharded harness: index `s` is the
    /// real loopback port this node's replica of shard group `s` binds. Empty for
    /// the single-group (`shards:` absent) harness — that path uses [`grpc`]
    /// verbatim and stays byte-for-byte the pre-m11p6 topology.
    shard_grpc: Vec<SocketAddr>,
}

/// A running single-region process.
struct NodeHandle {
    name: String,
    http: SocketAddr,
    process: Option<Child>,
}

impl NodeHandle {
    /// Send `kill -9` (SIGKILL) via the `kill` binary — a real crash, no graceful
    /// drain. Stays `unsafe_code = forbid`-clean (no `libc::kill` FFI), matching
    /// the SIGTERM path in `cluster_e2e.rs`. Reaps the process so it is not a
    /// zombie. Idempotent: a node already taken/killed is a no-op.
    fn sigkill(&mut self) {
        if let Some(mut child) = self.process.take() {
            #[cfg(unix)]
            let _ = Command::new("kill")
                .args(["-9", &child.id().to_string()])
                .status();
            #[cfg(not(unix))]
            let _ = child.kill();
            let _ = child.wait();
        }
    }

    /// Graceful SIGTERM-then-SIGKILL shutdown for Drop (same contract as
    /// `cluster_e2e.rs`): give the process 2s to checkpoint + WAL fsync + join,
    /// then hard-kill if it has not exited.
    fn graceful_stop(&mut self) {
        let Some(mut child) = self.process.take() else {
            return;
        };
        #[cfg(unix)]
        {
            let _ = Command::new("kill")
                .args(["-TERM", &child.id().to_string()])
                .status();
            let deadline = Instant::now() + Duration::from_secs(2);
            loop {
                match child.try_wait() {
                    Ok(Some(_)) => break,
                    _ if Instant::now() > deadline => {
                        let _ = child.kill();
                        break;
                    }
                    _ => std::thread::sleep(POLL_INTERVAL),
                }
            }
        }
        #[cfg(not(unix))]
        {
            let _ = child.kill();
        }
        let _ = child.wait();
    }
}

impl Drop for NodeHandle {
    fn drop(&mut self) {
        self.graceful_stop();
    }
}

/// The multi-process cluster harness: one OS process per region.
pub struct MultiProcCluster {
    plans: Vec<RegionPlan>,
    nodes: Vec<NodeHandle>,
    schema_path: PathBuf,
    /// Per-process topology files (index = region index). With the identity
    /// rewrite these are byte-identical; with a rewrite each lists real addrs for
    /// self and published addrs for peers.
    topology_paths: Vec<PathBuf>,
    rewrite: AddrRewrite,
    log: String,
    extra_env: HashMap<usize, Vec<(String, String)>>,
    client: reqwest::blocking::Client,
    /// Shard-group count (m11p6). `1` for the single-group harness (`shards:`
    /// absent); `S` for a [`start_sharded`](MultiProcCluster::start_sharded)
    /// cluster. Used by the per-shard status helpers.
    shards: usize,
    /// m12p4 partial placement: `placement[g]` = node indices hosting group `g`.
    /// `None` for full placement (every node hosts every group). Lets the
    /// per-shard leader wait know which node OWNS which group.
    partial_placement: Option<Vec<Vec<usize>>>,
    /// m12p4: how many shard groups each node hosts (index = node). Empty for the
    /// full-placement harnesses (every node hosts `shards`). The partial-aware
    /// leader wait uses this to size each node's expected `shards[]` row count.
    host_group_counts: Vec<usize>,
    /// Held so the tempdir (configs + per-node data dirs) outlives every process.
    _tmp: tempfile::TempDir,
}

impl MultiProcCluster {
    /// Spawn an `n`-region cluster with default options and wait for every
    /// process to become healthy. Region 0 is the leader.
    ///
    /// # Panics
    ///
    /// Panics if `n < 2`, if config/topology files cannot be written, if the
    /// binary cannot be built/spawned, or if any process does not become healthy
    /// within [`boot_budget`].
    #[must_use]
    pub fn start(n: usize) -> Self {
        Self::start_with(ClusterOptions::new(n))
    }

    /// Spawn a cluster with explicit options (extra env, rewrite hook, log level).
    ///
    /// # Panics
    ///
    /// See [`MultiProcCluster::start`].
    #[must_use]
    pub fn start_with(opts: ClusterOptions) -> Self {
        assert!(opts.regions >= 2, "need at least 2 regions for a cluster");
        // Serialize the spawn phase (build + port alloc + process start) across
        // test threads. Cargo runs the integration tests in this file in
        // parallel; several harnesses racing to `cargo build` the same binary and
        // claim ports at once is the only observed flakiness source, so the lock
        // covers ONLY the heavy bring-up. Once healthy, the tests run fully
        // concurrently against their own ports/tempdirs.
        let spawn_guard = spawn_lock()
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        let tmp = tempfile::tempdir().expect("create temp dir");
        let bin = tidal_server_bin();

        // Static plan: names, real ports, per-node data dirs (reused on restart).
        let plans: Vec<RegionPlan> = (0..opts.regions)
            .map(|i| {
                let data_dir = tmp.path().join(format!("region-{i}"));
                std::fs::create_dir_all(&data_dir).expect("create per-node data dir");
                RegionPlan {
                    name: region_name(i),
                    grpc: free_addr(),
                    http: free_addr(),
                    data_dir,
                    shard_grpc: Vec::new(),
                }
            })
            .collect();

        let schema_path = write_schema(tmp.path());
        let topology_paths: Vec<PathBuf> = (0..opts.regions)
            .map(|i| {
                write_topology_for(
                    tmp.path(),
                    &plans,
                    i,
                    &opts.rewrite,
                    opts.topology_extra.as_deref(),
                )
            })
            .collect();

        let mut harness = Self {
            plans,
            nodes: Vec::new(),
            schema_path,
            topology_paths,
            rewrite: opts.rewrite,
            log: opts.log,
            extra_env: opts.extra_env,
            client: reqwest::blocking::Client::builder()
                .build()
                .expect("build blocking client"),
            shards: 1,
            partial_placement: None,
            host_group_counts: Vec::new(),
            _tmp: tmp,
        };

        for i in 0..harness.plans.len() {
            let child = harness.spawn_process(&bin, i, &[]);
            let plan = &harness.plans[i];
            harness.nodes.push(NodeHandle {
                name: plan.name.clone(),
                http: plan.http,
                process: Some(child),
            });
        }

        std::thread::sleep(SETTLE);
        let deadline = Instant::now() + boot_budget();
        for i in 0..harness.nodes.len() {
            harness.wait_health_inner(i, deadline);
        }
        drop(spawn_guard);
        harness
    }

    /// Spawn an `nodes`-process cluster of `shards` shard groups at FULL placement
    /// (m11p6): every node replicates every group (RF = `nodes`), and group `s`'s
    /// term-0 leader is node `s` (balanced placement). `topology_extra` is appended
    /// to every per-process topology file (e.g. a fast-election `election:` block).
    ///
    /// Each `(node, shard)` binds its own real loopback gRPC port, so a node hosts
    /// `shards` `ShardReplica`s, each peering with the same group's replicas on the
    /// other nodes. This is the harness the m11p6 exit gate runs on: kill any node
    /// → only its shard-leaderships re-elect, the rest keep serving.
    ///
    /// # Panics
    ///
    /// Panics if `shards < 1`, `nodes < 2`, `shards > nodes` (balanced placement
    /// needs a distinct preferred-leader node per group), files cannot be written,
    /// or any process does not become healthy within [`boot_budget`].
    #[must_use]
    pub fn start_sharded(nodes: usize, shards: usize, topology_extra: Option<&str>) -> Self {
        assert!(nodes >= 2, "need at least 2 nodes for a cluster");
        assert!(shards >= 1, "need at least 1 shard group");
        assert!(
            shards <= nodes,
            "balanced placement needs a distinct preferred-leader node per group \
             (shards {shards} > nodes {nodes})"
        );
        let spawn_guard = spawn_lock()
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        let tmp = tempfile::tempdir().expect("create temp dir");
        let bin = tidal_server_bin();

        // Per node: one HTTP gateway port + one real gRPC port per shard group it
        // hosts (full placement ⇒ `shards` of them). `grpc` (the node base) is
        // shard 0's port, matching the topology `regions[].grpc_addr` convention.
        let plans: Vec<RegionPlan> = (0..nodes)
            .map(|i| {
                let data_dir = tmp.path().join(format!("region-{i}"));
                std::fs::create_dir_all(&data_dir).expect("create per-node data dir");
                let shard_grpc: Vec<SocketAddr> = (0..shards).map(|_| free_addr()).collect();
                RegionPlan {
                    name: region_name(i),
                    grpc: shard_grpc[0],
                    http: free_addr(),
                    data_dir,
                    shard_grpc,
                }
            })
            .collect();

        let schema_path = write_schema(tmp.path());
        let topology_paths: Vec<PathBuf> = (0..nodes)
            .map(|i| write_sharded_topology_for(tmp.path(), &plans, i, shards, topology_extra))
            .collect();

        let mut harness = Self {
            plans,
            nodes: Vec::new(),
            schema_path,
            topology_paths,
            rewrite: identity_rewrite(),
            log: "warn".into(),
            extra_env: HashMap::new(),
            client: reqwest::blocking::Client::builder()
                .build()
                .expect("build blocking client"),
            shards,
            partial_placement: None,
            host_group_counts: Vec::new(),
            _tmp: tmp,
        };

        for i in 0..harness.plans.len() {
            let child = harness.spawn_process(&bin, i, &[]);
            let plan = &harness.plans[i];
            harness.nodes.push(NodeHandle {
                name: plan.name.clone(),
                http: plan.http,
                process: Some(child),
            });
        }

        std::thread::sleep(SETTLE);
        let deadline = Instant::now() + boot_budget();
        for i in 0..harness.nodes.len() {
            harness.wait_health_inner(i, deadline);
        }
        drop(spawn_guard);
        harness
    }

    /// Spawn an `nodes`-process cluster of `placement.len()` shard groups at the
    /// EXPLICIT, possibly PARTIAL placement `placement` (m12p4 cross-shard read
    /// gate): `placement[g]` is the list of node indices that replicate group `g`,
    /// with `placement[g][0]` its term-0 preferred leader. A node that appears in
    /// some-but-not-all groups hosts a STRICT SUBSET of the corpus — exactly the
    /// shape the L4 cross-shard read fan-out exists for.
    ///
    /// Unlike [`start_sharded`](Self::start_sharded) (full placement, every node a
    /// replica of every group), here each node binds a gRPC port ONLY for the
    /// groups it actually hosts. The gateway on a partial node must fan a
    /// corpus-wide `/feed` out to the groups it does not host to be complete.
    ///
    /// # Panics
    ///
    /// Panics if `placement` is empty, any group lists no replica, a group's
    /// leader is not its first replica's owner, an index is out of range, files
    /// cannot be written, or any process is not healthy within [`boot_budget`].
    #[must_use]
    pub fn start_sharded_partial(
        nodes: usize,
        placement: &[Vec<usize>],
        topology_extra: Option<&str>,
    ) -> Self {
        assert!(nodes >= 2, "need at least 2 nodes for a cluster");
        assert!(!placement.is_empty(), "need at least 1 shard group");
        let shards = placement.len();
        for (g, replicas) in placement.iter().enumerate() {
            assert!(!replicas.is_empty(), "group {g} lists no replica");
            for &n in replicas {
                assert!(n < nodes, "group {g} names node {n} but only {nodes} nodes");
            }
        }
        let spawn_guard = spawn_lock()
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        let tmp = tempfile::tempdir().expect("create temp dir");
        let bin = tidal_server_bin();

        // Each node allocates one gRPC port PER GROUP IT HOSTS (not per group in
        // the cluster). `shard_grpc[g]` is bound only for hosted groups; a group
        // the node does not host gets a placeholder it never binds (its port is
        // never written into a `replicas:` entry, so nothing dials it).
        let host_groups: Vec<Vec<usize>> = (0..nodes)
            .map(|n| (0..shards).filter(|g| placement[*g].contains(&n)).collect())
            .collect();
        let plans: Vec<RegionPlan> = (0..nodes)
            .map(|i| {
                let data_dir = tmp.path().join(format!("region-{i}"));
                std::fs::create_dir_all(&data_dir).expect("create per-node data dir");
                // One real port per group; unhosted slots stay allocated but
                // unused (keeps `shard_grpc[g]` index-stable across all groups).
                let shard_grpc: Vec<SocketAddr> = (0..shards).map(|_| free_addr()).collect();
                RegionPlan {
                    name: region_name(i),
                    grpc: shard_grpc[0],
                    http: free_addr(),
                    data_dir,
                    shard_grpc,
                }
            })
            .collect();

        let schema_path = write_schema(tmp.path());
        let topology_paths: Vec<PathBuf> = (0..nodes)
            .map(|i| {
                write_partial_sharded_topology_for(tmp.path(), &plans, i, placement, topology_extra)
            })
            .collect();

        let harness = Self {
            plans,
            nodes: Vec::new(),
            schema_path,
            topology_paths,
            rewrite: identity_rewrite(),
            log: "warn".into(),
            extra_env: HashMap::new(),
            client: reqwest::blocking::Client::builder()
                .build()
                .expect("build blocking client"),
            shards,
            partial_placement: Some(placement.to_vec()),
            // Per-node hosted-group counts, so the partial-aware leader wait knows
            // how many `shards[]` rows each node should report.
            host_group_counts: host_groups.iter().map(Vec::len).collect(),
            _tmp: tmp,
        };
        let mut harness = harness;

        for i in 0..harness.plans.len() {
            let child = harness.spawn_process(&bin, i, &[]);
            let plan = &harness.plans[i];
            harness.nodes.push(NodeHandle {
                name: plan.name.clone(),
                http: plan.http,
                process: Some(child),
            });
        }

        std::thread::sleep(SETTLE);
        let deadline = Instant::now() + boot_budget();
        for i in 0..harness.nodes.len() {
            harness.wait_health_inner(i, deadline);
        }
        drop(spawn_guard);
        harness
    }

    /// Spawn one region's process. `extra` is appended to the per-node env
    /// (used by `restart` to inject overrides). The binary, schema, and this
    /// region's topology file are stable across restarts.
    fn spawn_process(&self, bin: &Path, idx: usize, extra: &[(String, String)]) -> Child {
        let plan = &self.plans[idx];
        let mut cmd = Command::new(bin);
        cmd.arg("cluster")
            .arg("--region")
            .arg(&plan.name)
            .arg("--listen")
            .arg(plan.http.to_string())
            .arg("--schema")
            .arg(&self.schema_path)
            .arg("--topology")
            .arg(&self.topology_paths[idx])
            .arg("--data-dir")
            .arg(&plan.data_dir)
            .env("TIDAL_ALLOW_EXPERIMENTAL_CLUSTER", "1")
            .env("TIDAL_SERVER_LOG", &self.log);
        // Node logs are discarded by default so parallel suites stay readable.
        // `Stdio::null()`, NEVER `Stdio::piped()`: a piped-but-undrained pipe
        // deadlocks the whole node once its 64KB kernel buffer fills — a
        // sufficiently chatty log path (e.g. ship-retry WARNs during a long
        // partition window) froze every logging thread in the leader,
        // including request handlers, and surfaced as spurious 408s.
        // `TIDAL_TEST_NODE_LOGS=inherit` streams them to the test's own
        // stderr for debugging a failing scenario.
        let inherit_logs = std::env::var("TIDAL_TEST_NODE_LOGS").is_ok_and(|v| v == "inherit");
        if inherit_logs {
            cmd.stdout(std::process::Stdio::inherit())
                .stderr(std::process::Stdio::inherit());
        } else {
            cmd.stdout(std::process::Stdio::null())
                .stderr(std::process::Stdio::null());
        }
        if let Some(env) = self.extra_env.get(&idx) {
            for (k, v) in env {
                cmd.env(k, v);
            }
        }
        for (k, v) in extra {
            cmd.env(k, v);
        }
        cmd.spawn()
            .unwrap_or_else(|e| panic!("failed to spawn tidal-server region {}: {e}", plan.name))
    }

    // ── Accessors ────────────────────────────────────────────────────────────

    /// This region's name (topology order).
    #[must_use]
    pub fn region_name(&self, idx: usize) -> &str {
        &self.plans[idx].name
    }

    /// All region names in topology id order.
    #[must_use]
    pub fn region_names(&self) -> Vec<String> {
        self.plans.iter().map(|p| p.name.clone()).collect()
    }

    /// Number of regions (= number of processes).
    #[must_use]
    pub const fn len(&self) -> usize {
        self.plans.len()
    }

    /// Whether the cluster has no regions (always false in practice; satisfies
    /// the clippy `len_without_is_empty` lint honestly).
    #[must_use]
    pub const fn is_empty(&self) -> bool {
        self.plans.is_empty()
    }

    /// The HTTP base URL (`http://host:port`) for a node's process.
    #[must_use]
    pub fn node(&self, idx: usize) -> String {
        format!("http://{}", self.plans[idx].http)
    }

    /// The shared blocking HTTP client (connection pooling across calls).
    #[must_use]
    pub const fn client(&self) -> &reqwest::blocking::Client {
        &self.client
    }

    /// Total on-disk size (bytes) of node `idx`'s data dir — the engine's
    /// recursive footprint after a snapshot install + catch-up. The exit-gate-2
    /// "≤5 min @ 100k items" evidence run records this as the installed-artifact
    /// proxy (a freshly seed-joined node's data dir IS the fetched snapshot plus
    /// the streamed suffix). Best-effort: unreadable entries are skipped.
    #[must_use]
    pub fn data_dir_bytes(&self, idx: usize) -> u64 {
        fn dir_bytes(path: &Path) -> u64 {
            let Ok(entries) = std::fs::read_dir(path) else {
                return 0;
            };
            entries
                .flatten()
                .map(|e| match e.file_type() {
                    Ok(ft) if ft.is_dir() => dir_bytes(&e.path()),
                    Ok(_) => e.metadata().map(|m| m.len()).unwrap_or(0),
                    Err(_) => 0,
                })
                .sum()
        }
        dir_bytes(&self.plans[idx].data_dir)
    }

    /// Node `idx`'s data directory (the `--data-dir` the process was spawned
    /// with). Used by the m12p6 SIGTERM test to assert the deterministic shutdown
    /// wrote its WAL checkpoint marker (`{data_dir}/wal/checkpoint.meta`) — the
    /// on-disk proof that `TidalDb::shutdown_inner` ran on the SIGTERM path
    /// (the same close that persists the HNSW graph).
    #[must_use]
    pub fn data_dir(&self, idx: usize) -> &Path {
        &self.plans[idx].data_dir
    }

    // ── HTTP helpers ─────────────────────────────────────────────────────────

    /// `GET {node(idx)}{path}`.
    #[must_use]
    pub fn get(&self, idx: usize, path: &str) -> reqwest::blocking::Response {
        let url = format!("{}{path}", self.node(idx));
        self.client
            .get(&url)
            .send()
            .unwrap_or_else(|e| panic!("GET {url} failed: {e}"))
    }

    /// `GET {node(idx)}{path}` returning the parsed JSON body (panics on failure).
    #[must_use]
    pub fn get_json(&self, idx: usize, path: &str) -> serde_json::Value {
        let resp = self.get(idx, path);
        let status = resp.status();
        resp.json()
            .unwrap_or_else(|e| panic!("GET {path} on node {idx} ({status}) not JSON: {e}"))
    }

    /// `POST {node(idx)}{path}` with a JSON body.
    #[must_use]
    pub fn post(
        &self,
        idx: usize,
        path: &str,
        body: &serde_json::Value,
    ) -> reqwest::blocking::Response {
        let url = format!("{}{path}", self.node(idx));
        self.client
            .post(&url)
            .json(body)
            .send()
            .unwrap_or_else(|e| panic!("POST {url} failed: {e}"))
    }

    /// `POST {node(idx)}{path}` carrying the `/sharded/*` single-copy opt-in
    /// header (`x-tidal-ack: local`).
    ///
    /// The `/sharded/*` WRITE routes apply to the owning region's local store with
    /// no WAL append, so they are single-copy regardless of the replication factor
    /// and now REJECT (400) a caller that has not said so explicitly. Every
    /// `/sharded/{items,embeddings,signals}` write in the suites goes through here
    /// so the opt-in is stated once.
    #[must_use]
    pub fn post_sharded(
        &self,
        idx: usize,
        path: &str,
        body: &serde_json::Value,
    ) -> reqwest::blocking::Response {
        let url = format!("{}{path}", self.node(idx));
        self.client
            .post(&url)
            .header("x-tidal-ack", "local")
            .json(body)
            .send()
            .unwrap_or_else(|e| panic!("POST {url} failed: {e}"))
    }

    // ── Crash / restart ──────────────────────────────────────────────────────

    /// SIGKILL the node at `idx` (a real crash — no graceful drain). After this,
    /// the node is DOWN; `wait_converged_all` / `wait_leader_agreed` skip it.
    ///
    /// # Panics
    ///
    /// Panics if `idx` is out of range.
    pub fn kill_hard(&mut self, idx: usize) {
        assert!(idx < self.nodes.len(), "kill_hard: node {idx} out of range");
        self.nodes[idx].sigkill();
    }

    /// Whether the node at `idx` is currently running (not killed / exited-on-Drop).
    #[must_use]
    pub fn is_alive(&self, idx: usize) -> bool {
        self.nodes[idx].process.is_some()
    }

    /// Restart a previously-killed node on the SAME ports + data dir, with extra
    /// env overrides (e.g. a new version tag, or an HLC skew). Reuses the region's
    /// topology file, so it rejoins the same cluster. Waits for it to become
    /// healthy before returning.
    ///
    /// # Panics
    ///
    /// Panics if `idx` is out of range or the restarted node does not become
    /// healthy within [`boot_budget`].
    /// Wipe a DEAD node's data dir (the operator "reseed" drill): the next
    /// `restart` boots it as a genuinely fresh node that recovers the full
    /// log via the catch-up stream. Panics if the process is still alive.
    pub fn wipe_data_dir(&self, idx: usize) {
        assert!(
            !self.is_alive(idx),
            "wipe_data_dir requires the node to be stopped"
        );
        let dir = &self.plans[idx].data_dir;
        std::fs::remove_dir_all(dir).expect("wipe data dir");
        std::fs::create_dir_all(dir).expect("recreate data dir");
    }

    pub fn restart(&mut self, idx: usize, env_overrides: &[(&str, &str)]) {
        assert!(idx < self.nodes.len(), "restart: node {idx} out of range");
        // Ensure any prior process is fully gone (idempotent if already killed).
        self.nodes[idx].sigkill();
        self.spawn_and_wait(idx, env_overrides);
    }

    /// Gracefully stop the node at `idx` with SIGTERM (the production shutdown
    /// path: the process flips readiness, drains in-flight requests, checkpoints,
    /// fsyncs the WAL, and joins its receiver before exiting — exactly what a
    /// rolling-upgrade SIGTERM gives an operator), waiting up to its graceful
    /// budget then hard-killing if it has not exited. After this the node is DOWN
    /// until [`restart`](Self::restart) brings it back on the same data dir. This
    /// is NOT [`kill_hard`](Self::kill_hard) (a crash with no drain) — the
    /// rolling-upgrade scenario must prove a CLEAN handoff with zero data loss, so
    /// it stops via SIGTERM and relies on WAL recovery only as a backstop.
    ///
    /// # Panics
    ///
    /// Panics if `idx` is out of range.
    pub fn stop_graceful(&mut self, idx: usize) {
        assert!(
            idx < self.nodes.len(),
            "stop_graceful: node {idx} out of range"
        );
        self.nodes[idx].graceful_stop();
    }

    /// Graceful rolling-upgrade restart: SIGTERM the running node (clean drain +
    /// checkpoint + WAL fsync), then bring it back on the SAME ports + data dir
    /// with `env_overrides` (e.g. a bumped `TIDAL_VERSION_TAG`). Reuses the
    /// region's topology file so it rejoins the same cluster, and waits for health
    /// before returning. This is the rolling-upgrade choreography's per-node step:
    /// no crash, a real shutdown signal, then WAL-backed recovery of pre-restart
    /// state.
    ///
    /// # Panics
    ///
    /// Panics if `idx` is out of range or the restarted node does not become
    /// healthy within [`boot_budget`].
    pub fn restart_graceful(&mut self, idx: usize, env_overrides: &[(&str, &str)]) {
        assert!(
            idx < self.nodes.len(),
            "restart_graceful: node {idx} out of range"
        );
        self.nodes[idx].graceful_stop();
        self.spawn_and_wait(idx, env_overrides);
    }

    /// Spawn `idx`'s process (under the spawn lock) with `env_overrides`, then wait
    /// for it to report healthy. Shared by [`restart`] and [`restart_graceful`];
    /// assumes the prior process is already stopped.
    fn spawn_and_wait(&mut self, idx: usize, env_overrides: &[(&str, &str)]) {
        let spawn_guard = spawn_lock()
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let bin = tidal_server_bin();
        let extra: Vec<(String, String)> = env_overrides
            .iter()
            .map(|(k, v)| ((*k).to_string(), (*v).to_string()))
            .collect();
        let child = self.spawn_process(&bin, idx, &extra);
        self.nodes[idx].process = Some(child);
        drop(spawn_guard);
        std::thread::sleep(SETTLE);
        let deadline = Instant::now() + boot_budget();
        self.wait_health_inner(idx, deadline);
    }

    // ── Elasticity: seed-join add / remove (m11p5 §3.4) ────────────────────────

    /// Add a NEW node that seed-joins the cluster (m11p5 §3.4): allocate ports +
    /// a data dir, write a knob-only topology variant (node `seed_of`'s topology
    /// file — the joiner reads its `regions:` list for KNOBS only, the join
    /// response is the roster), then spawn `cluster --region region-N
    /// --seed http://<seed http_addr> --advertise-grpc <p> --advertise-http <p2>
    /// --topology <knob file>`. Waits for `/health/startup` (the joiner reports
    /// 503 until first-converged, then 200 — so health gates on convergence,
    /// proving the snapshot+stream catch-up landed). Returns the new node's index.
    ///
    /// The data dir is a SUBDIRECTORY of the harness tempdir (a mount root would
    /// fail the §2.3 swap), so a snapshot install's rename siblings have a parent.
    ///
    /// # Panics
    ///
    /// Panics if `seed_of` is out of range, files cannot be written, the binary
    /// cannot be spawned, or the joiner does not become healthy within
    /// [`boot_budget`] (a join+install over loopback).
    pub fn add_node(&mut self, seed_of: usize) -> usize {
        assert!(
            seed_of < self.plans.len(),
            "add_node: seed {seed_of} out of range"
        );
        let spawn_guard = spawn_lock()
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        let idx = self.plans.len();
        let name = region_name(idx);
        let grpc = free_addr();
        let http = free_addr();
        // The harness tempdir is the parent of every node's data dir (the `_tmp`
        // field is underscore-held for Drop ordering; reach the path through an
        // existing plan to avoid touching the underscore binding directly).
        let tmp_root = self.plans[0]
            .data_dir
            .parent()
            .expect("a node data dir always has the tempdir as its parent")
            .to_path_buf();
        let data_dir = tmp_root.join(format!("region-{idx}"));
        std::fs::create_dir_all(&data_dir).expect("create joiner data dir");

        // The knob source (§3.5): a topology file whose `regions:` is the INITIAL
        // roster (NOT this joiner) — the joiner's own region is learned from the
        // join response, not this file. Reusing the seed node's topology file is
        // exactly the k8s shared-ConfigMap shape. We write a fresh copy so a
        // concurrent restart of the seed never races this file handle.
        let knob_path = {
            let path = tmp_root.join(format!("topology-seed-{idx}.yaml"));
            let body = std::fs::read_to_string(&self.topology_paths[seed_of])
                .expect("read seed topology for knobs");
            std::fs::write(&path, body).expect("write joiner knob topology");
            path
        };

        let seed_http = self.plans[seed_of].http;
        let bin = tidal_server_bin();
        let mut cmd = Command::new(&bin);
        cmd.arg("cluster")
            .arg("--region")
            .arg(&name)
            .arg("--listen")
            .arg(http.to_string())
            .arg("--schema")
            .arg(&self.schema_path)
            .arg("--topology")
            .arg(&knob_path)
            .arg("--data-dir")
            .arg(&data_dir)
            .arg("--seed")
            .arg(format!("http://{seed_http}"))
            .arg("--advertise-grpc")
            .arg(grpc.to_string())
            .arg("--advertise-http")
            .arg(http.to_string())
            .env("TIDAL_ALLOW_EXPERIMENTAL_CLUSTER", "1")
            .env("TIDAL_SERVER_LOG", &self.log)
            // Bound the join + the (rare) install handshake inside the test budget
            // rather than the production defaults.
            .env("TIDAL_SEED_JOIN_MS", "60000")
            .env("TIDAL_RESEED_HANDSHAKE_MS", "30000");
        let inherit_logs = std::env::var("TIDAL_TEST_NODE_LOGS").is_ok_and(|v| v == "inherit");
        if inherit_logs {
            cmd.stdout(std::process::Stdio::inherit())
                .stderr(std::process::Stdio::inherit());
        } else {
            cmd.stdout(std::process::Stdio::null())
                .stderr(std::process::Stdio::null());
        }
        let child = cmd
            .spawn()
            .unwrap_or_else(|e| panic!("failed to spawn seed-join node {name}: {e}"));

        // Register the new node in the harness tables (so `local_status`,
        // convergence polls, and Drop all see it).
        self.plans.push(RegionPlan {
            name: name.clone(),
            grpc,
            http,
            data_dir,
            shard_grpc: Vec::new(),
        });
        self.topology_paths.push(knob_path);
        self.nodes.push(NodeHandle {
            name,
            http,
            process: Some(child),
        });
        drop(spawn_guard);

        std::thread::sleep(SETTLE);
        // The joiner's /health/startup is 503 until first-converged (§4 sticky
        // readiness), so this wait proves the snapshot+stream catch-up landed.
        // Give it the convergence budget on top of the boot budget (a join +
        // possible install + catch-up).
        let deadline = Instant::now() + boot_budget() + convergence_budget();
        self.wait_health_inner(idx, deadline);
        idx
    }

    /// Remove (decommission) a member via the `/cluster/members/remove` verb
    /// (m11p5 §3.3): POST to the current leader (the verb forwards from a
    /// follower, but targeting the leader avoids a hop). After this the leader
    /// appends a `Removed` record; the removed node learns via the stream and
    /// flips to readiness 503 + stops campaigning. Returns the leader's response
    /// status code.
    ///
    /// # Panics
    ///
    /// Panics if no live leader can be found.
    pub fn remove_node(&self, name: &str) -> u16 {
        let leader_idx = self
            .live_indices()
            .into_iter()
            .find(|&i| {
                self.local_status(i)
                    .is_some_and(|s| s["is_leader"].as_bool() == Some(true))
            })
            .expect("remove_node: no live leader found to issue the remove verb");
        let resp = self.post(
            leader_idx,
            "/cluster/members/remove",
            &serde_json::json!({ "region": name }),
        );
        resp.status().as_u16()
    }

    // ── Health / convergence / leadership polling ──────────────────────────────

    /// Block until node `idx` reports healthy, or [`boot_budget`] elapses.
    ///
    /// # Panics
    ///
    /// Panics on timeout.
    pub fn wait_healthy(&self, idx: usize) {
        let deadline = Instant::now() + boot_budget();
        self.wait_health_inner(idx, deadline);
    }

    /// Poll `/health/startup` until 200, or `deadline`.
    fn wait_health_inner(&self, idx: usize, deadline: Instant) {
        let url = format!("{}/health/startup", self.node(idx));
        loop {
            if let Ok(resp) = self.client.get(&url).send()
                && resp.status().is_success()
            {
                return;
            }
            assert!(
                Instant::now() <= deadline,
                "node {idx} ({}) not healthy within budget: {}",
                self.plans[idx].name,
                self.drain_logs_hint(idx)
            );
            std::thread::sleep(POLL_INTERVAL);
        }
    }

    /// Block until EVERY live follower's OWN `/cluster/status/local` reports zero
    /// lag against the leader's high-water-mark, or `timeout`.
    ///
    /// This is the exact gap the old harness could not close: it polls each
    /// follower PROCESS directly (not the leader's aggregate view), so it proves
    /// the segment crossed the process boundary and was applied on the follower.
    /// The leader's own row is skipped (it writes directly, not via replication).
    /// A killed node (`process == None`) is skipped — a partition test converges
    /// only the reachable followers.
    ///
    /// # Panics
    ///
    /// Panics on timeout, dumping each lagging follower's last status row.
    pub fn wait_converged_all(&self, timeout: Duration) {
        let deadline = Instant::now() + timeout;
        loop {
            let leader_seq = self.leader_last_seq();
            let mut pending: Vec<String> = Vec::new();
            if let Some(target) = leader_seq {
                for idx in self.live_follower_indices() {
                    match self.local_status(idx) {
                        Some(st) => {
                            let applied = st["applied_events"].as_u64().unwrap_or(0);
                            let lag = st["lag_events"].as_u64().unwrap_or(u64::MAX);
                            if !(lag == 0 && applied >= target) {
                                pending.push(format!(
                                    "{}: applied={applied} lag={lag} (target {target})",
                                    self.plans[idx].name
                                ));
                            }
                        }
                        None => pending.push(format!(
                            "{}: status/local unreachable",
                            self.plans[idx].name
                        )),
                    }
                }
                if pending.is_empty() {
                    return;
                }
            } else {
                pending.push("leader status/local unreachable".into());
            }
            assert!(
                Instant::now() <= deadline,
                "cluster did not converge within {timeout:?}; pending: {pending:?}"
            );
            std::thread::sleep(POLL_INTERVAL);
        }
    }

    /// Block until every LIVE node's `/cluster/status/local` agrees the leader is
    /// `expected`, or `timeout`.
    ///
    /// # Panics
    ///
    /// Panics on timeout, dumping each live node's reported leader.
    pub fn wait_leader_agreed(&self, expected: &str, timeout: Duration) {
        let deadline = Instant::now() + timeout;
        loop {
            let mut disagree: Vec<String> = Vec::new();
            for idx in self.live_indices() {
                match self.local_status(idx) {
                    Some(st) => {
                        let leader = st["leader"].as_str().unwrap_or("<none>");
                        if leader != expected {
                            disagree.push(format!("{}: leader={leader}", self.plans[idx].name));
                        }
                    }
                    None => disagree.push(format!("{}: unreachable", self.plans[idx].name)),
                }
            }
            if disagree.is_empty() {
                return;
            }
            assert!(
                Instant::now() <= deadline,
                "nodes did not agree leader={expected} within {timeout:?}; disagree: {disagree:?}"
            );
            std::thread::sleep(POLL_INTERVAL);
        }
    }

    /// The leader's relay high-water-mark (`last_seq`) from its own
    /// `/cluster/status/local`, or `None` if the leader is down/unreachable.
    ///
    /// The leader is whichever LIVE node reports `is_leader: true` (so this works
    /// after a promote moves leadership).
    #[must_use]
    pub fn leader_last_seq(&self) -> Option<u64> {
        for idx in self.live_indices() {
            if let Some(st) = self.local_status(idx)
                && st["is_leader"].as_bool() == Some(true)
            {
                return st["last_seq"].as_u64();
            }
        }
        None
    }

    /// Fetch one node's `/cluster/status/local` (parsed JSON), or `None` on any
    /// transport/parse failure (a killed or partitioned node).
    #[must_use]
    pub fn local_status(&self, idx: usize) -> Option<serde_json::Value> {
        let url = format!("{}/cluster/status/local", self.node(idx));
        let resp = self
            .client
            .get(&url)
            .timeout(Duration::from_secs(2))
            .send()
            .ok()?;
        if !resp.status().is_success() {
            return None;
        }
        resp.json().ok()
    }

    /// Indices of nodes whose process is still running.
    fn live_indices(&self) -> Vec<usize> {
        (0..self.nodes.len())
            .filter(|&i| self.nodes[i].process.is_some())
            .collect()
    }

    /// Live nodes that are NOT the current leader (the convergence targets).
    fn live_follower_indices(&self) -> Vec<usize> {
        let leader = self.current_leader();
        self.live_indices()
            .into_iter()
            .filter(|&i| Some(self.plans[i].name.as_str()) != leader.as_deref())
            .collect()
    }

    /// The leader name as reported by the first live node that answers, or `None`.
    fn current_leader(&self) -> Option<String> {
        for idx in self.live_indices() {
            if let Some(st) = self.local_status(idx)
                && let Some(leader) = st["leader"].as_str()
            {
                return Some(leader.to_string());
            }
        }
        None
    }

    // ── Per-shard leadership (m11p6 sharded harness) ───────────────────────────

    /// The shard-group count this harness booted (1 for the single-group harness).
    #[must_use]
    pub const fn shard_count(&self) -> usize {
        self.shards
    }

    /// The leader EVERY live node agrees on for each shard group, or `None` if any
    /// live node disagrees, reports a null leader, or is unreachable. The
    /// agreed-leader read used to detect the converged steady state and to detect a
    /// failover (the map changes for exactly the killed node's groups).
    ///
    /// Reads the per-shard `shards[]` rows in `/cluster/status/local`: for each
    /// group it collects every live node's reported leader and accepts the group
    /// only when all of them are present and identical.
    #[must_use]
    pub fn agreed_shard_leaders(&self) -> Option<HashMap<u16, String>> {
        let live: Vec<usize> = self.live_indices();
        if live.is_empty() {
            return None;
        }
        let mut per_shard: HashMap<u16, String> = HashMap::new();
        for &idx in &live {
            let st = self.local_status(idx)?;
            let rows = st["shards"].as_array()?;
            // `shards[]` is one row per group this node HOSTS — a constant under
            // full placement (every node hosts every group), so a short array
            // means the node is still booting/unreachable (its status raced the
            // listener), not a leadership-convergence lag. Treating it as
            // not-yet-agreed (wait) is correct: a just-restarted rejoiner reports
            // a short/empty array until its groups open. (Partial placement would
            // make this per-node; the exit gate is full placement.)
            if rows.len() != self.shards {
                return None;
            }
            for row in rows {
                let shard = u16::try_from(row["shard"].as_u64()?).ok()?;
                let leader = row["leader"].as_str()?.to_string();
                match per_shard.get(&shard) {
                    Some(seen) if seen != &leader => return None, // disagreement
                    Some(_) => {}
                    None => {
                        per_shard.insert(shard, leader);
                    }
                }
            }
        }
        (per_shard.len() == self.shards).then_some(per_shard)
    }

    /// Block until every live node agrees on a (non-null) leader for every shard
    /// group, returning the agreed `shard -> leader` map. The sharded analogue of
    /// [`wait_leader_agreed`](Self::wait_leader_agreed).
    ///
    /// # Panics
    ///
    /// Panics on timeout, dumping the last partial view.
    #[must_use]
    pub fn wait_shard_leaders_agreed(&self, timeout: Duration) -> HashMap<u16, String> {
        let deadline = Instant::now() + timeout;
        loop {
            if let Some(map) = self.agreed_shard_leaders() {
                return map;
            }
            assert!(
                Instant::now() <= deadline,
                "shard leaders did not converge within {timeout:?}; last partial view: {:?}",
                self.agreed_shard_leaders()
            );
            std::thread::sleep(POLL_INTERVAL);
        }
    }

    /// m12p4 PARTIAL-placement analogue of [`agreed_shard_leaders`]. Each node
    /// reports only the `shards[]` rows for the groups IT hosts (a node-local
    /// count from `host_group_counts`), and a group is agreed once EVERY node that
    /// hosts it reports the SAME (non-null) leader. A node hosting a strict subset
    /// no longer trips the full-placement `rows.len() == self.shards` guard.
    ///
    /// # Panics
    ///
    /// Panics if called on a non-partial harness (use [`agreed_shard_leaders`]).
    #[must_use]
    fn agreed_shard_leaders_partial(&self) -> Option<HashMap<u16, String>> {
        let placement = self
            .partial_placement
            .as_ref()
            .expect("agreed_shard_leaders_partial needs a partial-placement harness");
        let live: Vec<usize> = self.live_indices();
        if live.is_empty() {
            return None;
        }
        let mut per_shard: HashMap<u16, String> = HashMap::new();
        for &idx in &live {
            let st = self.local_status(idx)?;
            let rows = st["shards"].as_array()?;
            // The node must have opened EVERY group it hosts (else it is still
            // booting — wait, do not mis-read as agreed).
            if rows.len() != self.host_group_counts[idx] {
                return None;
            }
            for row in rows {
                let shard = u16::try_from(row["shard"].as_u64()?).ok()?;
                let leader = row["leader"].as_str()?.to_string();
                match per_shard.get(&shard) {
                    Some(seen) if seen != &leader => return None,
                    Some(_) => {}
                    None => {
                        per_shard.insert(shard, leader);
                    }
                }
            }
        }
        // Agreed only when every group has a reported leader (some node hosts it).
        (per_shard.len() == placement.len()).then_some(per_shard)
    }

    /// Block until every group's hosting nodes agree on a leader, under PARTIAL
    /// placement. The m12p4 analogue of [`wait_shard_leaders_agreed`].
    ///
    /// # Panics
    ///
    /// Panics on timeout (dumping the last partial view) or on a non-partial
    /// harness.
    #[must_use]
    pub fn wait_shard_leaders_agreed_partial(&self, timeout: Duration) -> HashMap<u16, String> {
        let deadline = Instant::now() + timeout;
        loop {
            if let Some(map) = self.agreed_shard_leaders_partial() {
                return map;
            }
            assert!(
                Instant::now() <= deadline,
                "partial shard leaders did not converge within {timeout:?}; last view: {:?}",
                self.agreed_shard_leaders_partial()
            );
            std::thread::sleep(POLL_INTERVAL);
        }
    }

    /// Best-effort hint for a failing node, to make boot failures diagnosable in
    /// the panic message. Node logs go to /dev/null by default (re-run with
    /// `TIDAL_TEST_NODE_LOGS=inherit` to see them), so we surface the bind
    /// address — the actionable detail when a port conflicts or a bind fails.
    fn drain_logs_hint(&self, idx: usize) -> String {
        format!("(node http={})", self.plans[idx].http)
    }
}

// ── Shared seed / write helpers (the tier-3 suites' common data setup) ────────

/// Seed `1..=count` items (`title: "item {id}"`) plus deterministic 4-dim
/// embeddings (`[v, v+1, v+2, v+3]`, matching the shared schema's
/// `content_vector` slot) on the node at `leader_idx`.
///
/// m11p2: items and embeddings ride the one replicated log (kind-1/2 WAL
/// records) — there is no HTTP broadcast report to assert anymore. The 201 /
/// 204 assert leader durability (the record is fsynced into the stream);
/// follower visibility is asserted by each suite's convergence checks.
pub fn seed_items_and_embeddings(cluster: &MultiProcCluster, leader_idx: usize, count: u64) {
    for entity_id in 1..=count {
        let resp = cluster.post(
            leader_idx,
            "/items",
            &serde_json::json!({
                "entity_id": entity_id,
                "metadata": { "title": format!("item {entity_id}") }
            }),
        );
        assert_eq!(resp.status().as_u16(), 201, "leader /items must 201");

        // A deterministic, entity-varying embedding so feed ordering is stable.
        #[allow(clippy::cast_precision_loss)]
        let v = entity_id as f32;
        let resp = cluster.post(
            leader_idx,
            "/embeddings",
            &serde_json::json!({
                "entity_id": entity_id,
                "values": [v, v + 1.0, v + 2.0, v + 3.0]
            }),
        );
        assert_eq!(resp.status().as_u16(), 204, "leader /embeddings must 204");
    }
}

/// POST one `view` signal for `entity_id` to the node at `leader_idx`,
/// asserting the leader-durable 204 contract.
pub fn write_view(cluster: &MultiProcCluster, leader_idx: usize, entity_id: u64, weight: f64) {
    let resp = cluster.post(
        leader_idx,
        "/signals",
        &serde_json::json!({ "entity_id": entity_id, "signal": "view", "weight": weight }),
    );
    assert_eq!(
        resp.status().as_u16(),
        204,
        "leader /signals must 204 (leader-durable contract): {}",
        resp.status()
    );
}

// ── Topology + schema generation ──────────────────────────────────────────────

/// Write the topology file process `idx` will load. Its OWN region lists real
/// addrs (so it binds the harness-opened ports); every PEER lists the published
/// address from the rewrite hook.
///
/// The rewrite is invoked as `rewrite(observer_region, peer_region, kind, real)`:
/// the OBSERVER is `plans[idx]` (the process that will dial the published address)
/// and the PEER is `plans[i]` (the region it is reaching). Carrying the observer
/// lets a rewrite interpose a proxy on a SPECIFIC directed edge — the granularity
/// the follower↔follower partition test needs.
fn write_topology_for(
    dir: &Path,
    plans: &[RegionPlan],
    idx: usize,
    rewrite: &AddrRewrite,
    extra: Option<&str>,
) -> PathBuf {
    let path = dir.join(format!("topology-{idx}.yaml"));
    let observer = &plans[idx].name;
    let mut body = String::from("regions:\n");
    for (i, plan) in plans.iter().enumerate() {
        let (grpc, http) = if i == idx {
            // Self: real bind addresses, never rewritten.
            (plan.grpc.to_string(), plan.http.to_string())
        } else {
            (
                rewrite(observer, &plan.name, AddrKind::Grpc, plan.grpc),
                rewrite(observer, &plan.name, AddrKind::Http, plan.http),
            )
        };
        let _ = writeln!(body, "  - name: {}", plan.name);
        let _ = writeln!(body, "    grpc_addr: \"{grpc}\"");
        let _ = writeln!(body, "    http_addr: \"{http}\"");
    }
    // Region 0 is the initial leader (consistent with cluster_routes.rs).
    let _ = writeln!(body, "leader: {}", plans[0].name);
    if let Some(extra) = extra {
        let _ = writeln!(body, "{extra}");
    }
    let mut f = std::fs::File::create(&path).expect("create topology file");
    f.write_all(body.as_bytes()).expect("write topology file");
    path
}

/// Write the sharded topology file process `idx` loads (m11p6 full placement):
/// a `regions:` list (every node) plus a `shards:` block declaring `shards`
/// groups, each replicated across every node and led (term-0) by node `s`.
///
/// Each `(node, shard)` replica entry carries its own real loopback `grpc_addr`
/// (`plans[node].shard_grpc[shard]`), so the engine binds a distinct socket per
/// hosted group — no derived-port collisions. With identity addressing every
/// process's file is byte-identical (the kill-node exit gate uses SIGKILL, not
/// TCP-proxy partitions, so no per-observer rewrite is needed). The legacy
/// top-level `leader:` is set but unused once `shards:` is present.
fn write_sharded_topology_for(
    dir: &Path,
    plans: &[RegionPlan],
    idx: usize,
    shards: usize,
    extra: Option<&str>,
) -> PathBuf {
    // Full placement: balanced leaders need a distinct node per group, and every
    // node must have one bound port per group. Assert both up front so a future
    // refactor that desyncs `shards` from the allocation fails loudly here, not as
    // an out-of-bounds index mid-write or an opaque engine bind error.
    assert!(
        shards <= plans.len(),
        "balanced placement needs nodes ({}) >= shards ({shards})",
        plans.len()
    );
    for plan in plans {
        assert!(
            plan.shard_grpc.len() >= shards,
            "node {} has {} shard ports, need {shards}",
            plan.name,
            plan.shard_grpc.len()
        );
    }
    let path = dir.join(format!("topology-{idx}.yaml"));
    let mut body = String::from("regions:\n");
    for plan in plans {
        // The node's base grpc_addr is shard 0's port (the topology convention);
        // each group sets its own explicit grpc_addr in the `shards:` block below.
        let _ = writeln!(body, "  - name: {}", plan.name);
        let _ = writeln!(body, "    grpc_addr: \"{}\"", plan.shard_grpc[0]);
        let _ = writeln!(body, "    http_addr: \"{}\"", plan.http);
    }
    let _ = writeln!(body, "shards:");
    for s in 0..shards {
        let _ = writeln!(body, "  - id: {s}");
        // Balanced placement: group `s`'s term-0 leader is node `s`.
        let _ = writeln!(body, "    leader: {}", plans[s].name);
        let _ = writeln!(body, "    replicas:");
        for plan in plans {
            let _ = writeln!(
                body,
                "      - {{ node: {}, grpc_addr: \"{}\" }}",
                plan.name, plan.shard_grpc[s]
            );
        }
    }
    // Legacy field — required by the loader, ignored once `shards:` is present.
    let _ = writeln!(body, "leader: {}", plans[0].name);
    if let Some(extra) = extra {
        let _ = writeln!(body, "{extra}");
    }
    let mut f = std::fs::File::create(&path).expect("create sharded topology file");
    f.write_all(body.as_bytes())
        .expect("write sharded topology file");
    path
}

/// Write the PARTIAL-placement sharded topology file process `idx` loads (m12p4):
/// a `regions:` list (every node) plus a `shards:` block where each group `g`
/// declares ONLY the replica entries in `placement[g]` (a subset of nodes), led
/// (term-0) by `placement[g][0]`. A node absent from `placement[g]` never appears
/// in group `g`'s `replicas:`, so it does not host the group — its gateway must
/// cross-shard fan out to reach `g`. Identity addressing, so every process's file
/// is byte-identical (SIGKILL not TCP-proxy, no per-observer rewrite).
fn write_partial_sharded_topology_for(
    dir: &Path,
    plans: &[RegionPlan],
    idx: usize,
    placement: &[Vec<usize>],
    extra: Option<&str>,
) -> PathBuf {
    let path = dir.join(format!("topology-{idx}.yaml"));
    let mut body = String::from("regions:\n");
    for plan in plans {
        let _ = writeln!(body, "  - name: {}", plan.name);
        let _ = writeln!(body, "    grpc_addr: \"{}\"", plan.shard_grpc[0]);
        let _ = writeln!(body, "    http_addr: \"{}\"", plan.http);
    }
    let _ = writeln!(body, "shards:");
    for (g, replicas) in placement.iter().enumerate() {
        assert!(!replicas.is_empty(), "group {g} lists no replica");
        let _ = writeln!(body, "  - id: {g}");
        // Term-0 leader = the group's first listed replica node.
        let _ = writeln!(body, "    leader: {}", plans[replicas[0]].name);
        let _ = writeln!(body, "    replicas:");
        for &node in replicas {
            let _ = writeln!(
                body,
                "      - {{ node: {}, grpc_addr: \"{}\" }}",
                plans[node].name, plans[node].shard_grpc[g]
            );
        }
    }
    // Legacy field — required by the loader, ignored once `shards:` is present.
    let _ = writeln!(body, "leader: {}", plans[0].name);
    if let Some(extra) = extra {
        let _ = writeln!(body, "{extra}");
    }
    let mut f = std::fs::File::create(&path).expect("create partial sharded topology file");
    f.write_all(body.as_bytes())
        .expect("write partial sharded topology file");
    path
}

/// Write the shared schema file. Matches the signal set the in-process route
/// tests use (`view` decayed, `like` decayed, `hide` permanent for `/hardnegs`)
/// so the harness asserts against identical engine behavior, plus a `title` text
/// field and a 4-dim embedding slot so `/items` metadata and `/embeddings` apply.
fn write_schema(dir: &Path) -> PathBuf {
    let path = dir.join("schema.yaml");
    std::fs::write(
        &path,
        r"signals:
  - name: view
    entity: item
    decay:
      exponential:
        half_life_seconds: 604800
    windows: [one_hour]
    velocity: false
  - name: like
    entity: item
    decay:
      exponential:
        half_life_seconds: 86400
    windows: [one_hour]
    velocity: false
  - name: hide
    entity: item
    decay:
      permanent: true
    velocity: false
text_fields:
  - name: title
    kind: text
embedding_slots:
  - name: content_vector
    entity: item
    dimensions: 4
",
    )
    .expect("write schema.yaml");
    path
}

/// Region name for index `i`: a fixed roster (`us-east`, `eu-west`, `ap-south`,
/// …) matching the in-process route tests, then `region-N` past the roster.
fn region_name(i: usize) -> String {
    const ROSTER: [&str; 3] = ["us-east", "eu-west", "ap-south"];
    ROSTER
        .get(i)
        .map_or_else(|| format!("region-{i}"), |s| (*s).to_string())
}

// ── Process-global helpers (shared across harness instances) ────────────────────

/// Allocate a free loopback port by binding `:0` and reading the assigned port.
/// The listener is dropped immediately; the brief TOCTOU window is acceptable
/// for a localhost test harness. The spawn lock serializes allocation WITHIN
/// one test binary; suites in different binaries (cargo runs each `tests/*.rs`
/// as its own process) can still race, and a killed node's port could be handed
/// to another suite before `restart` reclaims it. Both manifest as a loud bind
/// failure at boot/restart — never silent corruption — and the ephemeral range
/// makes them rare enough to accept over a cross-process lock file.
fn free_addr() -> SocketAddr {
    TcpListener::bind("127.0.0.1:0")
        .expect("bind ephemeral port")
        .local_addr()
        .expect("read local_addr")
}

/// Serialize the heavy spawn phase (build + port alloc + process start) across
/// the parallel integration tests in a file. See [`MultiProcCluster::start_with`].
fn spawn_lock() -> &'static Mutex<()> {
    static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
    LOCK.get_or_init(|| Mutex::new(()))
}

/// Resolve (and build once) the `tidal-server` binary, exactly like
/// `cluster_e2e.rs`. The `cargo build` is a no-op when the binary is current.
fn tidal_server_bin() -> PathBuf {
    static BIN: OnceLock<PathBuf> = OnceLock::new();
    BIN.get_or_init(|| {
        let manifest_dir = Path::new(env!("CARGO_MANIFEST_DIR"));
        let candidates = [
            manifest_dir.parent(),
            manifest_dir.parent().and_then(Path::parent),
        ];
        let workspace_root = candidates
            .into_iter()
            .flatten()
            .find(|root| root.join("target").is_dir() || root.join("Cargo.toml").is_file())
            .or_else(|| manifest_dir.parent())
            .expect("locate workspace root");

        // Build with `fault-injection` so the spawned cluster processes carry the
        // m11p9 WAL slow-fsync / disk-full hooks (inert unless a `TIDAL_FAULT_*`
        // env var arms them — every non-fault suite spawns the same binary and is
        // unaffected). Production never passes this feature, so the shipped image
        // compiles the hooks out entirely.
        // stderr is CAPTURED, not discarded. It used to be `Stdio::null()`, and on
        // 2026-08-31 this assert fired in CI with the message below and nothing else —
        // no compiler error, no ENOSPC, no OOM, nothing to act on. A build failure
        // whose reason is thrown away costs more than the build. The output is only
        // read on the failure path, so a passing run is unchanged.
        let out = Command::new("cargo")
            .arg("build")
            .arg("-p")
            .arg("tidal-server")
            .arg("--features")
            .arg("fault-injection")
            .current_dir(workspace_root)
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::piped())
            .output()
            .expect("run cargo build");
        assert!(
            out.status.success(),
            "cargo build -p tidal-server --features fault-injection failed \
             (exit {:?}) in {}:\n{}",
            out.status.code(),
            workspace_root.display(),
            String::from_utf8_lossy(&out.stderr)
        );

        let bin = workspace_root.join("target/debug/tidal-server");
        assert!(
            bin.exists(),
            "tidal-server binary not found at {}",
            bin.display()
        );
        bin
    })
    .clone()
}