//! Tier-3 sharding × replication suite (m11p6 L4, REAL multi-process cluster).
//!
//! The m11p6 exit gate, run over real OS processes — 3 shard groups × RF=3, every
//! node a replica of every group, each group its own elected leader (balanced:
//! group `s` led by node `s`). Two pillars:
//!
//! 1. **Kill-node failover localizes to the dead node's groups** — SIGKILL a node
//!    under concurrent `ack=quorum` load and prove: (a) ONLY the groups that node
//!    led re-elect a new leader (the groups led by survivors keep theirs); (b) the
//!    re-election completes inside the failover budget (<10s); (c) READS NEVER STOP
//!    (a concurrent `/feed` poller on a survivor sees zero failures across the
//!    window — reads are served from local replicas, no leader needed); and
//!    (d) ZERO acknowledged-write loss per group — every item the client saw a
//!    2xx + `x-tidal-seq` for is present afterwards on its shard's NEW leader (the
//!    m11p4 vote restriction guarantees the elected leader holds every committed
//!    write).
//!
//! 2. **The L3 rebalance verbs move exactly one group** — `POST
//!    /cluster/shards/{id}/transfer` and `/cluster/promote?shard=` move one
//!    group's leadership and leave the others untouched, and `POST
//!    /cluster/shards/{id}/replicas` (remove) runs a per-group fenced conf-change.
//!
//! Kill-point count: `TIDAL_SHARDING_KILLPOINTS` (default 3 for CI; the recorded
//! exit-gate run sweeps more — see docs/planning/milestone-11/phase-6.md).
//!
//! Run: `cargo test -p tidal-server --features cluster-e2e --test cluster_sharding -- --nocapture`

#![cfg(feature = "cluster-e2e")]
#![allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::panic,
    clippy::cast_possible_truncation,
    clippy::cast_precision_loss,
    clippy::too_many_lines
)]

mod support;

use std::collections::BTreeMap;
use std::sync::{
    Arc,
    atomic::{AtomicBool, AtomicU64, Ordering},
};
use std::time::{Duration, Instant};

use support::{
    invariants::{AckLedger, item_token, post_acked},
    multiproc::{MultiProcCluster, convergence_budget},
};
use tidaldb::{replication::shard::ShardRouter, schema::EntityId};

/// Fast election (mirrors `cluster_election.rs`): heartbeat 100ms, timeout
/// 500–1000ms, lease 350ms — so a SIGKILL failover completes well inside the
/// budget. Applied to every group (each `ShardReplica` reads this block).
const FAST_ELECTION_YAML: &str = "election:\n  heartbeat_interval_ms: 100\n  election_timeout_min_ms: 500\n  election_timeout_max_ms: 1000\n  leader_lease_ms: 350";

/// "Failover < 10s p99" — the roadmap's m11p6 bound (also p4's).
const FAILOVER_BUDGET: Duration = Duration::from_secs(10);

const NODES: usize = 3;
const SHARDS: usize = 3;

/// CI-default kill points; the exit-gate run sets `TIDAL_SHARDING_KILLPOINTS` higher.
fn killpoints() -> usize {
    std::env::var("TIDAL_SHARDING_KILLPOINTS")
        .ok()
        .and_then(|v| v.parse().ok())
        .filter(|&n| n > 0)
        .unwrap_or(3)
}

/// Serializes the heavy multi-process tests in THIS target.
///
/// Each test here spawns 3 OS processes hosting 3 shard groups each. The harness's
/// `spawn_lock` only serializes the spawn itself and is released as soon as
/// `start_sharded` returns, so without this every test in the file can have a live
/// cluster simultaneously — 4 clusters, 12 processes, all electing and shipping at
/// once. That contention makes `mp_sharded_kill_node_moves_only_its_leaderships_zero_loss`
/// miss its 10s failover budget: it passes alone and fails in the full target.
///
/// Held for the whole test body, so exactly one cluster is alive at a time. Poison is
/// recovered rather than propagated: one failing test must not cascade into "the rest
/// panicked on a poisoned lock", which hides the original failure.
fn heavy_test_guard() -> std::sync::MutexGuard<'static, ()> {
    static LOCK: std::sync::OnceLock<std::sync::Mutex<()>> = std::sync::OnceLock::new();
    LOCK.get_or_init(|| std::sync::Mutex::new(()))
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
}

/// Find the node index whose region name matches `name`.
fn idx_of(cluster: &MultiProcCluster, name: &str) -> usize {
    (0..cluster.len())
        .find(|&i| cluster.region_name(i) == name)
        .unwrap_or_else(|| panic!("no node named {name}"))
}

/// The m11p6 headline exit gate over real OS processes: kill a node under
/// `ack=quorum` load and prove only ITS shard-leaderships move (<10s), reads
/// never stop, and zero acknowledged writes are lost — across several kill points.
#[test]
fn mp_sharded_kill_node_moves_only_its_leaderships_zero_loss() {
    let _heavy = heavy_test_guard();
    let mut cluster = MultiProcCluster::start_sharded(NODES, SHARDS, Some(FAST_ELECTION_YAML));
    // The gateway's entity→shard hash (the same FNV-1a router every node routes by).
    let router = ShardRouter::hash(SHARDS as u16).expect("build shard router");

    // Balanced placement: each group `s` is led (term 0) by node `s`.
    let initial = cluster.wait_shard_leaders_agreed(convergence_budget());
    for s in 0..SHARDS as u16 {
        assert_eq!(
            initial[&s],
            cluster.region_name(usize::from(s)),
            "group {s} should boot led by node {s} (balanced placement)"
        );
    }

    let mut entity: u64 = 1;
    for round in 0..killpoints() {
        // Wait for a fully-converged steady state (all groups agree their leader).
        let before = cluster.wait_shard_leaders_agreed(convergence_budget());
        // Target the leader of group `round % SHARDS` — "kill any node".
        let target_shard = (round % SHARDS) as u16;
        let target_name = before[&target_shard].clone();
        let target_idx = idx_of(&cluster, &target_name);
        // The groups this node currently leads (its leaderships must move) and the
        // rest (must stay put — the localization property).
        let led: Vec<u16> = (0..SHARDS as u16)
            .filter(|s| before[s] == target_name)
            .collect();
        let others: Vec<u16> = (0..SHARDS as u16)
            .filter(|s| before[s] != target_name)
            .collect();
        // Write + read through a SURVIVOR gateway so the client's entry node never
        // dies mid-request (it forwards group-leader writes; reads serve locally).
        let survivor = (0..NODES)
            .find(|&i| i != target_idx)
            .expect("a live survivor gateway distinct from the kill target");
        let survivor_base = cluster.node(survivor);

        // ── Concurrent reads that must never stop ─────────────────────────────
        let stop = Arc::new(AtomicBool::new(false));
        let read_failures = Arc::new(AtomicU64::new(0));
        let read_ok = Arc::new(AtomicU64::new(0));
        let reader = {
            let stop = Arc::clone(&stop);
            let fails = Arc::clone(&read_failures);
            let oks = Arc::clone(&read_ok);
            let base = survivor_base.clone();
            std::thread::spawn(move || {
                let client = reqwest::blocking::Client::builder()
                    .timeout(Duration::from_secs(3))
                    .build()
                    .unwrap();
                while !stop.load(Ordering::Acquire) {
                    match client
                        .get(format!("{base}/feed?profile=for_you&limit=24"))
                        .send()
                    {
                        Ok(r) if r.status().is_success() => {
                            oks.fetch_add(1, Ordering::Relaxed);
                        }
                        _ => {
                            fails.fetch_add(1, Ordering::Relaxed);
                        }
                    }
                    std::thread::sleep(Duration::from_millis(50));
                }
            })
        };

        // ── Two writer threads: ack=quorum items+views, entities spread across
        //    all groups by the gateway hash. Record only what the client saw acked.
        let mut writers = Vec::new();
        for w in 0..2u64 {
            let stop = Arc::clone(&stop);
            let base = survivor_base.clone();
            let first = entity + w * 100_000;
            writers.push(std::thread::spawn(move || {
                let client = reqwest::blocking::Client::builder()
                    .timeout(Duration::from_secs(3))
                    .build()
                    .unwrap();
                let mut acked: Vec<(u64, u64, Option<u64>)> = Vec::new();
                let mut e = first;
                while !stop.load(Ordering::Acquire) {
                    let item_seq = post_acked(
                        &client,
                        &base,
                        "/items",
                        "quorum",
                        &serde_json::json!({
                            "entity_id": e, "metadata": { "title": item_token(e) }
                        }),
                    );
                    let view_seq = post_acked(
                        &client,
                        &base,
                        "/signals",
                        "quorum",
                        &serde_json::json!({ "entity_id": e, "signal": "view", "weight": 1.0 }),
                    );
                    if let Some(s) = item_seq {
                        acked.push((e, s, view_seq));
                    }
                    e += 1;
                }
                acked
            }));
        }

        // Pseudo-random kill point per round (reproducible — no Math.random here).
        std::thread::sleep(Duration::from_millis(200 + (round as u64 * 131) % 500));
        cluster.kill_hard(target_idx);
        stop.store(true, Ordering::Release);

        // ── Collect the acked ledger and the per-shard coverage. ──────────────
        let mut ledger = AckLedger::new();
        let mut per_shard: BTreeMap<u16, usize> = BTreeMap::new();
        for wj in writers {
            for (e, item_seq, view_seq) in wj.join().expect("writer thread") {
                let s = router.route(EntityId::new(e));
                *per_shard.entry(s.0).or_default() += 1;
                ledger.record(e, item_seq, view_seq);
            }
        }
        reader.join().expect("reader thread");

        // ── (a)+(b) Failover localizes to the killed node's groups, <budget. ──
        let deadline = Instant::now() + FAILOVER_BUDGET;
        let after = loop {
            if let Some(now) = cluster.agreed_shard_leaders() {
                let led_moved = led
                    .iter()
                    .all(|s| now.get(s).is_some_and(|l| l != &target_name));
                let others_stable = others.iter().all(|s| now.get(s) == before.get(s));
                if led_moved && others_stable {
                    break now;
                }
            }
            assert!(
                Instant::now() <= deadline,
                "round {round}: killing {target_name} (led {led:?}) did not localize within \
                 {FAILOVER_BUDGET:?}: before={before:?}, now={:?}",
                cluster.agreed_shard_leaders()
            );
            std::thread::sleep(Duration::from_millis(100));
        };
        for s in &others {
            assert_eq!(
                after.get(s),
                before.get(s),
                "round {round}: group {s} (led by a survivor) must NOT change leader"
            );
        }

        // ── (c) Reads never stopped. ──────────────────────────────────────────
        assert_eq!(
            read_failures.load(Ordering::Relaxed),
            0,
            "round {round}: {} feed reads on survivor {survivor} FAILED during failover \
             (reads must never stop); {} succeeded",
            read_failures.load(Ordering::Relaxed),
            read_ok.load(Ordering::Relaxed)
        );
        assert!(
            read_ok.load(Ordering::Relaxed) > 0,
            "round {round}: the read poller never got a single 2xx — it was not exercising reads"
        );

        // A round with ZERO acked writes proves nothing — the failover would have
        // had no acknowledged state to lose. The writers run hundreds of quorum
        // writes in the pre-kill window, so an empty ledger means a setup fault
        // (writers never got a 2xx), not a passing round. The ledger spans every
        // group the gateway hash routed to (logged per-shard); the loss proof
        // below probes each acked write on ITS shard's new leader, so coverage of
        // a killed node's groups is whatever genuinely routed there this round.
        assert!(
            !ledger.is_empty(),
            "round {round}: no acked writes recorded — the kill tested nothing \
             (writers saw 0 quorum acks before the kill at {target_name})"
        );
        println!(
            "round {round}: killed {target_name} (led {led:?}); failover -> {after:?}; \
             {} acked writes, per-shard {per_shard:?}; reads {} ok / 0 failed",
            ledger.len(),
            read_ok.load(Ordering::Relaxed)
        );

        // ── (d) ZERO acknowledged-write loss: every acked item is present on ITS
        //    shard's NEW leader. A quorum ack means the write committed (a majority
        //    held it durably), and the m11p4 vote restriction guarantees the
        //    elected leader holds every committed write — so the post-failover
        //    shard leader is the authoritative place to prove presence (a still
        //    catching-up follower is not). Poll past the text index's 2s
        //    auto-commit (the leader scatter-reads all its hosted groups).
        for w in ledger.writes() {
            let shard = router.route(EntityId::new(w.entity_id)).0;
            let leader_name = after.get(&shard).expect("post-failover leader for shard");
            let leader_idx = idx_of(&cluster, leader_name);
            assert!(
                item_present(&cluster, leader_idx, w.entity_id, Duration::from_secs(15)),
                "round {round} (killed {target_name}): ACKNOWLEDGED LOSS — item {} \
                 (seq {}, shard {shard}, acked) is missing on shard {shard}'s new leader \
                 {leader_name} (node {leader_idx})",
                w.entity_id,
                w.item_seq
            );
        }

        // Restart the killed node so the cluster is whole for the next kill point
        // (one node down at a time keeps every group's quorum intact). The doubled
        // budget covers BOTH the rejoiner's boot+catch-up AND any group whose
        // leadership is still settling — stacked, not just convergence.
        entity += 10_000;
        cluster.restart(target_idx, &[]);
        let _ = cluster.wait_shard_leaders_agreed(convergence_budget() + convergence_budget());
    }
}

/// The L3 rebalance verbs over real processes: `POST /cluster/shards/{id}/transfer`
/// and `/cluster/promote?shard=` move EXACTLY one group's leadership, and the
/// per-group remove verb runs a fenced conf-change — proving `?shard=` selection
/// and the per-group reuse of the m11p4/m11p5 machinery end to end.
#[test]
fn mp_sharded_rebalance_verbs_move_one_group() {
    let _heavy = heavy_test_guard();
    let cluster = MultiProcCluster::start_sharded(NODES, SHARDS, Some(FAST_ELECTION_YAML));
    let before = cluster.wait_shard_leaders_agreed(convergence_budget());
    for s in 0..SHARDS as u16 {
        assert_eq!(before[&s], cluster.region_name(usize::from(s)));
    }
    let node0 = cluster.region_name(0).to_string();
    let node1 = cluster.region_name(1).to_string();

    // ── Transfer group 0's leadership node0 -> node1 (RESTful rebalance verb). ──
    let resp = cluster.post(
        0,
        "/cluster/shards/0/transfer",
        &serde_json::json!({ "region": node1 }),
    );
    assert!(
        resp.status().is_success(),
        "shards/0/transfer must succeed, got {}",
        resp.status()
    );
    wait_until(FAILOVER_BUDGET, || {
        cluster
            .agreed_shard_leaders()
            .is_some_and(|m| m[&0] == node1 && m[&1] == before[&1] && m[&2] == before[&2])
    });
    let mid = cluster
        .agreed_shard_leaders()
        .expect("agreed after transfer");
    assert_eq!(
        mid[&0], node1,
        "group 0 leadership must have moved to node1"
    );
    assert_eq!(mid[&1], before[&1], "group 1 leadership must be untouched");
    assert_eq!(mid[&2], before[&2], "group 2 leadership must be untouched");

    // ── Move it back with the shard-scoped promote (`?shard=` selector). ──────
    let resp = cluster.post(
        0,
        "/cluster/promote?shard=0",
        &serde_json::json!({ "region": node0 }),
    );
    assert!(
        resp.status().is_success(),
        "promote?shard=0 must succeed, got {}",
        resp.status()
    );
    wait_until(FAILOVER_BUDGET, || {
        cluster
            .agreed_shard_leaders()
            .is_some_and(|m| m[&0] == node0)
    });

    // ── The per-group replica verb is wired + `?shard=`-scoped: assert its input
    //    validation and that `/cluster/members?shard=` selects the right group's
    //    roster. (The live add/remove conf-change reuses the m11p5 machinery
    //    per-group — proven by the membership suite — but a node's readiness is
    //    still node-global across its co-hosted groups, so removing a node from
    //    ONE group of a multi-group node is a tracked S>1 follow-up, NOT asserted
    //    here. See docs/planning/milestone-11/phase-6.md.)
    let bad_action = cluster.post(
        0,
        "/cluster/shards/0/replicas",
        &serde_json::json!({ "action": "frobnicate", "name": node1 }),
    );
    assert_eq!(
        bad_action.status().as_u16(),
        400,
        "an unknown shard-replica action must be a 400"
    );
    let add_missing_addrs = cluster.post(
        0,
        "/cluster/shards/0/replicas",
        &serde_json::json!({ "action": "add", "name": "region-9" }),
    );
    assert_eq!(
        add_missing_addrs.status().as_u16(),
        400,
        "add without grpc_addr/http_addr must be a 400"
    );
    // The `?shard=` selector resolves a hosted group's roster (full placement ⇒
    // every group lists all three nodes); an unhosted shard id is a 400.
    let roster = cluster.get_json(0, "/cluster/members?shard=2");
    assert_eq!(
        roster["members"].as_array().map(Vec::len),
        Some(NODES),
        "group 2's roster must list every node under full placement"
    );
    let unhosted = cluster.post(
        0,
        "/cluster/shards/9/transfer",
        &serde_json::json!({ "region": node0 }),
    );
    assert_eq!(
        unhosted.status().as_u16(),
        400,
        "targeting a shard this node does not host must be a 400"
    );
}

/// Whether item `entity` is searchable on node `idx` within `budget` — the
/// content presence probe (`/search?query=<token>`), polling past the text
/// index's ~2s auto-commit. A clean 2xx-with-no-hit through the whole budget is
/// genuine absence (returns false); transport/non-2xx is retried until the
/// budget, so a just-promoted leader still warming up is not charged as absence
/// prematurely.
fn item_present(cluster: &MultiProcCluster, idx: usize, entity: u64, budget: Duration) -> bool {
    let token = item_token(entity);
    let base = cluster.node(idx);
    let client = reqwest::blocking::Client::builder()
        .timeout(Duration::from_secs(4))
        .build()
        .unwrap();
    let deadline = Instant::now() + budget;
    loop {
        if let Ok(resp) = client
            .get(format!("{base}/search?query={token}&limit=5"))
            .send()
            && resp.status().is_success()
        {
            let body: serde_json::Value = resp.json().unwrap_or(serde_json::Value::Null);
            let hit = body["items"].as_array().is_some_and(|items| {
                items
                    .iter()
                    .any(|it| it["entity_id"].as_u64() == Some(entity))
            });
            if hit {
                return true;
            }
        }
        if Instant::now() > deadline {
            return false;
        }
        std::thread::sleep(Duration::from_millis(200));
    }
}

/// Poll `cond` every 100ms until it returns true or `budget` elapses; returns the
/// final value of `cond` (so a caller can assert it true with context).
fn wait_until(budget: Duration, mut cond: impl FnMut() -> bool) -> bool {
    let deadline = Instant::now() + budget;
    loop {
        if cond() {
            return true;
        }
        if Instant::now() > deadline {
            return false;
        }
        std::thread::sleep(Duration::from_millis(100));
    }
}

/// An embedding written through ANY node must be searchable on EVERY replica of
/// its shard group, WITHOUT a restart.
///
/// Guards the replication contract of the NON-sharded write surface (`/items`,
/// `/embeddings`): those ride the leader WAL relay, so every replica of the entity's
/// group must end up able to answer for it.
///
/// Uses `/items` + `/embeddings` DELIBERATELY, not `/sharded/*`. The `/sharded/*`
/// surface hash-partitions and applies to the owner's LOCAL store without a WAL
/// append (`node.rs:8828-8829`, `ShardReplica::apply_embedding_local`), so it is
/// single-copy BY DESIGN and a parity assertion against it fails correctly — see
/// `mp_sharded_surface_writes_are_local_to_the_owner` below, which pins that
/// intended behavior. Writing a parity test against the wrong surface is exactly
/// the mistake that made a 2026-08-30 live probe look like a durability incident.
///
/// **The assertion is deliberately PRE-restart.** `rebuild_from_store` at open
/// re-derives the whole index from the durable store, so a test that restarts before
/// asserting converges the replicas regardless of whether the live path works — it
/// would have PASSED against the bug it exists to catch. That is exactly how the
/// live divergence stayed hidden across a rolling deploy.
///
/// Full placement (3 groups on every node) means `/vector_search` scatters over all
/// locally hosted groups and is corpus-complete per node, so "absent here" is a real
/// absence and not a fan-out artifact.
#[test]
fn mp_embedding_is_searchable_on_every_replica_without_restart() {
    let _heavy = heavy_test_guard();
    let cluster = MultiProcCluster::start_sharded(NODES, SHARDS, Some(FAST_ELECTION_YAML));
    let leaders = cluster.wait_shard_leaders_agreed(convergence_budget());
    println!("[parity] group leaders: {leaders:?}");

    // Entity ids are hash-routed across groups, so a spread covers all three and
    // the test does not depend on which group any single id lands in.
    const ENTITIES: [u64; 6] = [911_001, 911_002, 911_003, 911_004, 911_005, 911_006];

    // Write each entity through a DIFFERENT node, round-robin. The live failure was
    // sensitive to which node accepted the write (it never landed on the acceptor),
    // so exercising every entry point is the point.
    for (n, entity) in ENTITIES.iter().enumerate() {
        let via = n % NODES;
        let resp = cluster.post(
            via,
            "/items",
            &serde_json::json!({ "entity_id": entity, "metadata": { "title": "parity" } }),
        );
        assert_eq!(
            resp.status().as_u16(),
            201,
            "entity {entity} via node {via}: /items must 201"
        );
        let resp = cluster.post(
            via,
            "/embeddings",
            &serde_json::json!({ "entity_id": entity, "values": embedding_for(*entity) }),
        );
        assert_eq!(
            resp.status().as_u16(),
            204,
            "entity {entity} via node {via}: /embeddings must 204"
        );
    }

    cluster.wait_converged_all(convergence_budget());

    // Poll every node for every entity, bounded by the convergence budget. Replication
    // is asynchronous, so a brief absence is legitimate; a PERSISTENT one is the bug.
    let deadline = Instant::now() + convergence_budget();
    let mut missing: Vec<(usize, u64)> = Vec::new();
    loop {
        missing.clear();
        for entity in ENTITIES {
            for node in 0..NODES {
                if !vector_search_finds(&cluster, node, entity) {
                    missing.push((node, entity));
                }
            }
        }
        if missing.is_empty() || Instant::now() >= deadline {
            break;
        }
        std::thread::sleep(Duration::from_millis(250));
    }

    if !missing.is_empty() {
        // Dump the blob ledger before failing: `originated` on the writer vs
        // `applied`/`apply_failed` per peer localises the gap to enqueue, ship, or
        // apply. Diagnosing from the failure output beats re-running by hand.
        for node in 0..NODES {
            println!(
                "[parity] node {node} ({}) blob ledger:\n{}",
                cluster.region_name(node),
                blob_ledger(&cluster, node)
            );
        }
    }
    assert!(
        missing.is_empty(),
        "an embedding must be searchable on EVERY replica without a restart, but \
         (node, entity) pairs are still missing after the convergence budget: {missing:?}. \
         The non-sharded surface rides the leader WAL relay, so this means a blob left \
         the leader and never landed - check the blob ledger dumped above."
    );
}

/// This node's `tidaldb_cluster_blobs_*` and per-group vector-count lines.
fn blob_ledger(cluster: &MultiProcCluster, node: usize) -> String {
    let resp = cluster.get(node, "/metrics");
    if resp.status().as_u16() != 200 {
        return format!("  <metrics unavailable: HTTP {}>", resp.status());
    }
    resp.text().map_or_else(
        |e| format!("  <metrics body unreadable: {e}>"),
        |body| {
            body.lines()
                .filter(|l| {
                    (l.starts_with("tidaldb_cluster_blobs_")
                        || l.starts_with("tidaldb_usearch_vector_count"))
                        && !l.ends_with(" 0")
                })
                .map(|l| format!("  {l}"))
                .collect::<Vec<_>>()
                .join("\n")
        },
    )
}

/// A deterministic, entity-varying 4-dim embedding.
///
/// NOT a constant vector: every constant vector normalizes to the same unit vector,
/// so all of them are equidistant and a presence test over them proves nothing.
fn embedding_for(entity: u64) -> Vec<f32> {
    #[allow(clippy::cast_precision_loss)]
    let v = (entity % 997) as f32;
    vec![v, v + 1.0, v + 2.0, v + 3.0]
}

/// Does `node`'s own `/vector_search` return `entity` for its exact stored vector?
///
/// Queries the node's OWN address, never a shared/load-balanced endpoint — routing a
/// probe through a gateway hides precisely this class of bug (it answers from an
/// arbitrary replica).
fn vector_search_finds(cluster: &MultiProcCluster, node: usize, entity: u64) -> bool {
    let resp = cluster.post(
        node,
        "/vector_search",
        &serde_json::json!({ "vector": embedding_for(entity), "k": 5 }),
    );
    if resp.status().as_u16() != 200 {
        return false;
    }
    let body: serde_json::Value = match resp.json() {
        Ok(b) => b,
        Err(_) => return false,
    };
    body["items"].as_array().is_some_and(|items| {
        items
            .iter()
            .any(|it| it["entity_id"].as_u64() == Some(entity))
    })
}

/// The `/sharded/*` write surface is single-copy BY DESIGN, and now says so — pin
/// both halves.
///
/// `node.rs:8828-8829` states it: the `/sharded/*` surface hash-partitions and applies
/// to the owning region's LOCAL store, and does NOT ride the leader WAL relay (that is
/// the non-sharded surface). `sharded_write_embedding` therefore calls
/// `ShardReplica::apply_embedding_local`, which performs no WAL append and so ships
/// nothing to peers.
///
/// This test exists because that property is easy to mistake for a replication defect:
/// a 2026-08-30 live probe wrote embeddings through `/sharded/embeddings`, found each
/// one on exactly one of three nodes, and was initially recorded as a durability
/// incident. It is not one — but a caller who assumes `/sharded/*` writes are
/// replicated is building on sand, so the semantics deserve an executable statement
/// rather than a comment.
///
/// If a future change makes `/sharded/*` replicate, this test SHOULD fail: that is a
/// deliberate contract change, and the failure is the prompt to update the docs, the
/// runbook, and any durability claim that depends on it.
///
/// Since the opt-in landed the surface REQUIRES `x-tidal-ack: local`, so this test
/// pins two things: the single-copy semantics WHEN opted in, and the 400 (naming
/// the header and the replicating alternative) when not. Making the tradeoff
/// explicit is what stops a caller taking it by accident; it does not change what
/// the tradeoff is.
#[test]
fn mp_sharded_surface_writes_are_local_to_the_owner() {
    let _heavy = heavy_test_guard();
    let cluster = MultiProcCluster::start_sharded(NODES, SHARDS, Some(FAST_ELECTION_YAML));
    let _leaders = cluster.wait_shard_leaders_agreed(convergence_budget());

    // Six ids spread across the hash space, each written through a different node.
    //
    // The writes carry `x-tidal-ack: local` (via `post_sharded`) because the
    // surface now REQUIRES that opt-in. The point of this test is unchanged: the
    // surface is still single-copy WHEN opted in. The opt-in makes the tradeoff a
    // decision; it does not soften it.
    const ENTITIES: [u64; 6] = [922_001, 922_002, 922_003, 922_004, 922_005, 922_006];
    for (n, entity) in ENTITIES.iter().enumerate() {
        let via = n % NODES;
        assert_eq!(
            cluster
                .post_sharded(
                    via,
                    "/sharded/items",
                    &serde_json::json!({ "entity_id": entity, "metadata": {} })
                )
                .status()
                .as_u16(),
            201
        );
        assert_eq!(
            cluster
                .post_sharded(
                    via,
                    "/sharded/embeddings",
                    &serde_json::json!({ "entity_id": entity, "values": embedding_for(*entity) })
                )
                .status()
                .as_u16(),
            204
        );
    }

    // And WITHOUT the opt-in the same write is refused, with a body that names the
    // header and the replicating alternative. This is the half that makes the
    // single-copy tradeoff impossible to take by accident.
    let refused = cluster.post(
        0,
        "/sharded/items",
        &serde_json::json!({ "entity_id": 922_099_u64, "metadata": {} }),
    );
    assert_eq!(
        refused.status().as_u16(),
        400,
        "a /sharded/* write with no x-tidal-ack: local opt-in must be refused"
    );
    let body = refused.text().unwrap_or_default();
    for expected in ["x-tidal-ack", "local", "SINGLE-COPY", "/items"] {
        assert!(
            body.contains(expected),
            "the rejection must name {expected:?}: {body}"
        );
    }
    cluster.wait_converged_all(convergence_budget());
    // Generous settle: the claim is "never replicates", so give replication every
    // chance to happen before asserting that it did not.
    std::thread::sleep(Duration::from_secs(5));

    // At least one entity must be visible on strictly fewer than all replicas.
    // Asserting "exactly 1 node" for EVERY entity would be over-fitting: reads scatter
    // over all locally hosted groups, so an owner that co-locates the querying group
    // can legitimately answer for more than one id.
    let spread: Vec<(u64, usize)> = ENTITIES
        .iter()
        .map(|&e| {
            (
                e,
                (0..NODES)
                    .filter(|&n| vector_search_finds(&cluster, n, e))
                    .count(),
            )
        })
        .collect();
    println!("[sharded-semantics] visible-replica counts: {spread:?}");
    assert!(
        spread.iter().any(|&(_, n)| n < NODES),
        "the /sharded/* surface must NOT replicate (node.rs:8828-8829); every entity \
         reached all {NODES} replicas, so the contract changed: {spread:?}"
    );
    assert!(
        spread.iter().all(|&(_, n)| n >= 1),
        "every /sharded/* write must still be durable on its owner: {spread:?}"
    );
}

/// The shared title token every seeded item carries, so ONE `/search?query=` hit
/// returns the whole seeded page (`item_token` is per-entity unique and would
/// return exactly one row, which cannot show a rank sequence).
const RANK_TOKEN: &str = "rnkdense";

/// Entity ids spread across the hash space so all three groups contribute rows
/// on the multi-group shape. 12 ⇒ several rows per group, so a per-group counter
/// that was never re-stamped shows up as duplicates rather than by luck.
const RANK_ENTITIES: [u64; 12] = [
    933_001, 933_002, 933_003, 933_004, 933_005, 933_006, 933_007, 933_008, 933_009, 933_010,
    933_011, 933_012,
];

/// One ranked page from `node`, as `(rank, score)` in wire order.
fn ranked_page(cluster: &MultiProcCluster, node: usize, path: &str) -> Vec<(u64, f64)> {
    let body = cluster.get_json(node, path);
    body["items"]
        .as_array()
        .unwrap_or(&Vec::new())
        .iter()
        .map(|it| {
            (
                it["rank"].as_u64().unwrap_or_default(),
                it["score"].as_f64().unwrap_or_default(),
            )
        })
        .collect()
}

/// Seed `RANK_ENTITIES` on a fresh `shards`-group cluster and assert every node's
/// `/feed` and `/search` page carries a DENSE ascending `rank` with descending
/// scores.
fn assert_dense_ranks(shards: usize) {
    let cluster = MultiProcCluster::start_sharded(NODES, shards, Some(FAST_ELECTION_YAML));
    let _leaders = cluster.wait_shard_leaders_agreed(convergence_budget());

    for (n, entity) in RANK_ENTITIES.iter().enumerate() {
        let via = n % NODES;
        let resp = cluster.post(
            via,
            "/items",
            &serde_json::json!({
                "entity_id": entity,
                "metadata": { "title": format!("{RANK_TOKEN} {}", item_token(*entity)) }
            }),
        );
        assert_eq!(
            resp.status().as_u16(),
            201,
            "entity {entity} via node {via}: /items must 201"
        );
    }
    cluster.wait_converged_all(convergence_budget());

    let limit = RANK_ENTITIES.len();
    let feed_path = format!("/feed?profile=for_you&limit={limit}");
    let search_path = format!("/search?query={RANK_TOKEN}&limit={limit}");

    // The text index auto-commits on a ~2s cadence, so poll until the whole
    // seeded page is visible on every node. A SHORT page would make the density
    // assertion pass vacuously (a 1-row page is trivially dense).
    for path in [&feed_path, &search_path] {
        let full = wait_until(convergence_budget(), || {
            (0..NODES).all(|n| ranked_page(&cluster, n, path).len() == limit)
        });
        assert!(
            full,
            "shards={shards}: every node must serve all {limit} seeded rows on {path} before \
             rank density can be judged; got {:?}",
            (0..NODES)
                .map(|n| ranked_page(&cluster, n, path).len())
                .collect::<Vec<_>>()
        );
    }

    for node in 0..NODES {
        for path in [&feed_path, &search_path] {
            let page = ranked_page(&cluster, node, path);
            let ranks: Vec<u64> = page.iter().map(|&(r, _)| r).collect();
            let expected: Vec<u64> = (1..=page.len() as u64).collect();
            assert_eq!(
                ranks, expected,
                "shards={shards} node {node} {path}: rank must be a dense ascending 1..n \
                 sequence. Duplicates are the signature of per-group slices merged without a \
                 re-stamp (scatter_merge); zeros mean a remote slice was never stamped."
            );
            for w in page.windows(2) {
                assert!(
                    w[1].1 <= w[0].1 + 1e-9,
                    "shards={shards} node {node} {path}: scores must stay descending — the \
                     rank stamp is not a re-sort. Got {:?} then {:?}",
                    w[0],
                    w[1]
                );
            }
        }
    }
    println!("[rank] shards={shards}: dense 1..{limit} on /feed and /search, all {NODES} nodes");
}

/// `rank` on a cluster's ranked reads must be DENSE and ascending — on BOTH of
/// the two structurally different paths the same handler reaches the wire by.
///
/// Under full placement (production: every group on every node) `missing_groups()`
/// is empty and `/feed` / `/search` return straight out of `scatter_merge`
/// (`node.rs`), skipping `merge_cross_shard` — the sibling that DOES re-stamp.
/// Live `/search?query=verification` returned ranks `1,1,2`: each hosted group
/// ranks its own slice `1..k` locally, the slices get concatenated and
/// score-sorted, and nothing renumbered them. `scatter_merge` now takes the same
/// `set_rank` closure and stamps after `truncate`.
///
/// Two shapes, because there are two paths:
///  - **multi-group** (`shards = SHARDS`): the fan-out + merge path, where the
///    stamp lives. This is the production shape and the one that was broken.
///  - **single-group** (`shards = 1`): the `[only]` fast path returns the engine's
///    own result and deliberately BYPASSES the stamp. That the engine already
///    ranks densely is an assumption until asserted, so assert it.
///
/// Scores are checked to stay descending on both: the fix is a renumbering, never
/// a re-sort, so an ordering change here would be a different and worse defect.
#[test]
fn mp_ranked_reads_stamp_dense_rank_on_both_merge_shapes() {
    let _heavy = heavy_test_guard();
    assert_dense_ranks(SHARDS);
    assert_dense_ranks(1);
}