//! Tier-3 MULTI-PROCESS partition-injection chaos suite (m8p10 task 05).
//!
//! These tests inject REAL network partitions between independent OS processes
//! using an in-harness TCP relay ([`support::partition::PartitionProxy`]), then
//! assert the leader-ships replication fabric behaves correctly under and after a
//! genuine transport-layer severance — not an engine flag.
//!
//! # Why a TCP proxy and not iptables/pfctl
//!
//! The ROADMAP sanctions "iptables/pfctl OR a proxy (toxiproxy or similar)".
//! `iptables`/`pfctl` need root and diverge between Linux and macOS, so a developer
//! laptop cannot run them unprivileged. A user-space TCP relay the harness owns is
//! identical on every platform, root-free, and severs at the genuine OS transport
//! layer: established streams are `shutdown(Both)` (the leader's in-flight gRPC
//! `ShipSegment` returns a real transport error and the HTTP fan-out times out) and
//! new connects are accepted-then-dropped (a fresh dial fails its HTTP/2 handshake).
//! That is a true partition between two processes. The proxy interposes ONLY on the
//! path PEERS use to reach a region (the published topology address); the test
//! client always talks to each node's REAL HTTP address directly, so the operator's
//! console survives the partition — exactly the runbook drill's "read the stale
//! follower" step. NO test here ever touches `/cluster/partition` (the simulated
//! ship-skip flag); the chaos is all real sockets.
//!
//! # What each test proves
//!
//! * `mp_uat_step3_degraded_query_during_partition` — under a real partition of one
//!   region, leader writes still 204 (leader-durable contract), the partitioned
//!   region's applied STALLS while the leader's high-water-mark climbs (lag is real,
//!   observed via direct addrs), the leader's aggregated `/cluster/status` reports
//!   the region `reachable: false` with worst-case lag, `GET /sharded/feed` degrades
//!   honestly (`degraded: true`, names the unavailable shard) yet still returns live
//!   results, and the partitioned region's OWN `/feed` serves its PRE-partition data
//!   (eventual consistency, not an outage).
//! * `mp_uat_step4_heal_reconcile_no_loss_no_dup` — healing the link + `/cluster/heal`
//!   redelivers every missed segment with ZERO loss (applied == leader `last_seq`) and
//!   ZERO duplication (feed-score parity to 1e-6); a SECOND heal is idempotent (state
//!   unchanged); `/cluster/reconcile` converges a hard-negative recorded during the
//!   window ("hides remain hidden" verified via `/feed?user_id=` on the reconciled
//!   follower) and reports merge+apply < 100ms BOTH sides; a repeat reconcile leaves
//!   the converged state identical AND the scores EXACTLY unchanged (bug 2 fixed: the
//!   CRDT signal merge is idempotent, so reconcile of converged nodes is a fixpoint —
//!   previously the scores crept 0.5 → 0.375 → … and this was unassertable).
//! * `mp_partition_between_followers` — the ROADMAP "partition between two followers"
//!   criterion, made explicit and HONEST: severing only the eu-west↔ap-south directed
//!   edge does NOT impede correctness (no replication traffic flows follower↔follower
//!   in a leader-ships topology, so both followers keep converging via their intact
//!   leader links). Then ALSO severing the leader→ap-south edge shows the contrast —
//!   lag climbs ONLY then — and a heal restores convergence with no data loss.
//!
//! # Budget
//!
//! Tier-3 over real OS processes. Three tests, each: ~boot (shared binary build,
//! amortized by `cargo build` no-op) + a bounded partition window + a bounded
//! convergence wait. Every wait is poll-with-deadline; there are no bare sleeps used
//! as a correctness gate. Whole-suite wall budget < 4 minutes on a developer laptop
//! (each test's individual waits sum well under 60s).
//!
//! ```bash
//! cargo test -p tidal-server --features cluster-e2e --test cluster_chaos -- --nocapture
//! ```
#![cfg(feature = "cluster-e2e")]
// Tier-3 harness allows, mirroring `cluster_multiproc.rs` / `cluster_routes.rs`:
// `unwrap` on known-good fixtures is idiomatic test noise; the lossy numeric casts
// are the same pervasive-and-intentional scoring math the crate config documents.
#![allow(
    clippy::unwrap_used,
    clippy::missing_panics_doc,
    clippy::too_many_lines,
    clippy::cast_precision_loss,
    clippy::cast_possible_truncation,
    clippy::cast_sign_loss,
    clippy::items_after_statements
)]

mod support;

use std::time::{Duration, Instant};

use support::{
    multiproc::{
        BREAKER_RESET, ClusterOptions, MultiProcCluster, convergence_budget,
        seed_items_and_embeddings, write_view,
    },
    partition::proxied_rewrite,
};

/// Region 0 = `us-east` = the initial leader in every topology.
const LEADER: usize = 0;
/// Region 1 = `eu-west` (a follower, intact in the UAT 3/4 partitions).
const EU_WEST: usize = 1;
/// Region 2 = `ap-south` = the region we partition in UAT steps 3/4.
const AP_SOUTH: usize = 2;

/// A short partition window during which the leader keeps writing.
const PARTITION_WRITES: u64 = 8;

// ── Shared seeding / feed helpers (canonical bodies in support::multiproc) ──────

/// A node's local-region feed as a sorted `(entity_id, score)` vector.
fn feed_pairs(
    cluster: &MultiProcCluster,
    idx: usize,
    profile: &str,
    limit: u32,
) -> Vec<(u64, f64)> {
    let body = cluster.get_json(idx, &format!("/feed?profile={profile}&limit={limit}"));
    let mut pairs: Vec<(u64, f64)> = body["items"]
        .as_array()
        .unwrap_or(&Vec::new())
        .iter()
        .map(|it| {
            (
                it["entity_id"].as_u64().unwrap(),
                it["score"].as_f64().unwrap(),
            )
        })
        .collect();
    pairs.sort_by_key(|(id, _)| *id);
    pairs
}

/// A user-scoped feed's item-id set (sorted). Hard negatives for `user_id` are
/// filtered by the engine's user-context stage, so a hidden item is ABSENT here.
fn feed_item_ids_for_user(
    cluster: &MultiProcCluster,
    idx: usize,
    user_id: u64,
    profile: &str,
    limit: u32,
) -> Vec<u64> {
    let body = cluster.get_json(
        idx,
        &format!("/feed?profile={profile}&limit={limit}&user_id={user_id}"),
    );
    let mut ids: Vec<u64> = body["items"]
        .as_array()
        .unwrap_or(&Vec::new())
        .iter()
        .map(|it| it["entity_id"].as_u64().unwrap())
        .collect();
    ids.sort_unstable();
    ids
}

/// Assert two feed views carry the SAME items with scores equal to 1e-6.
fn assert_feed_parity(label: &str, a: &[(u64, f64)], b: &[(u64, f64)]) {
    assert_eq!(
        a.iter().map(|(id, _)| *id).collect::<Vec<_>>(),
        b.iter().map(|(id, _)| *id).collect::<Vec<_>>(),
        "{label}: feed item sets differ"
    );
    for ((id_a, score_a), (_, score_b)) in a.iter().zip(b.iter()) {
        assert!(
            (score_a - score_b).abs() <= 1e-6,
            "{label}: score for item {id_a} differs: {score_a} vs {score_b}"
        );
    }
}

/// The leader's high-water-mark (`last_seq`), polled until non-`None`.
fn leader_seq(cluster: &MultiProcCluster) -> u64 {
    cluster
        .leader_last_seq()
        .expect("leader status/local reachable")
}

/// One node's `applied_events` from its OWN `/cluster/status/local` (direct addr,
/// which survives the partition because the test client never uses the proxy).
fn applied(cluster: &MultiProcCluster, idx: usize) -> u64 {
    cluster
        .local_status(idx)
        .and_then(|st| st["applied_events"].as_u64())
        .unwrap_or(0)
}

/// Poll `pred()` until true or `budget` elapses, asserting with `msg` on timeout.
fn poll_until(budget: Duration, msg: &str, mut pred: impl FnMut() -> bool) {
    let deadline = Instant::now() + budget;
    while !pred() {
        assert!(Instant::now() <= deadline, "timed out: {msg}");
        std::thread::sleep(Duration::from_millis(50));
    }
}

// The leader-ships circuit breaker ([`BREAKER_RESET`], the `tidal-net` default)
// opens after 5 consecutive failed ships and stays open for 30s. After a heal,
// the leader's transport cannot reach the recovered peer until the breaker
// half-opens and one probe ship closes it — so a SINGLE `/cluster/heal` issued
// before that window is a no-op (the ships hit the open breaker). This is
// genuine production behavior, not a flag: the operator re-runs the heal (or a
// new write fires the eager-ship probe) until the breaker recovers. Tests bound
// post-heal convergence at BREAKER_RESET + the convergence budget, re-issuing
// `/cluster/heal` so the first heal after the breaker half-opens redelivers
// every missed batch.

/// True when every live follower has applied up to (or past) the leader's
/// high-water-mark with zero lag. Non-panicking (unlike `wait_converged_all`), so a
/// retry loop can poll it.
fn converged(cluster: &MultiProcCluster, followers: &[usize]) -> bool {
    let Some(target) = cluster.leader_last_seq() else {
        return false;
    };
    followers.iter().all(|&idx| {
        cluster.local_status(idx).is_some_and(|st| {
            let applied = st["applied_events"].as_u64().unwrap_or(0);
            let lag = st["lag_events"].as_u64().unwrap_or(u64::MAX);
            lag == 0 && applied >= target
        })
    })
}

/// Heal `region`'s link and drive convergence the way an operator does: re-issue
/// `POST /cluster/heal` until every `followers` entry converges, allowing for the
/// circuit-breaker reset window. Asserts on timeout.
fn heal_until_converged(cluster: &MultiProcCluster, region: &str, followers: &[usize]) {
    let deadline = Instant::now() + BREAKER_RESET + convergence_budget();
    loop {
        let resp = cluster.post(
            LEADER,
            "/cluster/heal",
            &serde_json::json!({ "region": region }),
        );
        assert_eq!(
            resp.status().as_u16(),
            200,
            "/cluster/heal on leader must 200: {}",
            resp.status()
        );
        // Give the just-issued redelivery a moment to apply, then check. If the
        // breaker is still open this is a no-op and we loop; once it half-opens the
        // probe ship closes it and this heal redelivers everything.
        let check_deadline = Instant::now() + Duration::from_secs(3);
        while Instant::now() <= check_deadline {
            if converged(cluster, followers) {
                return;
            }
            std::thread::sleep(Duration::from_millis(100));
        }
        assert!(
            Instant::now() <= deadline,
            "region '{region}' did not converge within breaker-reset + convergence budget; \
             leader hwm={:?}, follower applied={:?}",
            cluster.leader_last_seq(),
            followers
                .iter()
                .map(|&i| applied(cluster, i))
                .collect::<Vec<_>>()
        );
    }
}

// ── UAT step 3: degraded global query during a real partition ───────────────────

#[test]
fn mp_uat_step3_degraded_query_during_partition() {
    // Interpose TCP proxies on every inbound edge of ap-south. Peers reach
    // ap-south through the proxy; ap-south self-binds its real ports; the test
    // client talks to every node's REAL http addr directly.
    let (rewrite, proxies) = proxied_rewrite(&["ap-south"]);
    let cluster = MultiProcCluster::start_with(ClusterOptions::new(3).with_rewrite(rewrite));

    const ITEMS: u64 = 12;
    seed_items_and_embeddings(&cluster, LEADER, ITEMS);
    for entity_id in 1..=ITEMS {
        write_view(&cluster, LEADER, entity_id, entity_id as f64);
    }
    cluster.wait_converged_all(convergence_budget());

    // ap-south's pre-partition feed (served from its own process, direct addr).
    let pre_partition_feed = feed_pairs(&cluster, AP_SOUTH, "trending", ITEMS as u32);
    assert!(
        !pre_partition_feed.is_empty(),
        "ap-south must rank the seeded items before the partition"
    );
    let applied_at_partition = applied(&cluster, AP_SOUTH);

    // ── SEVER: a real TCP partition of ap-south from every peer ────────────────
    proxies.region("ap-south").sever_all();
    println!("[step3] severed ap-south (gRPC + HTTP) from all peers");

    // Leader keeps writing. Each must STILL 204 — the leader durably applies and
    // best-effort ships; the ship to ap-south fails engine-side (WARN), but the
    // write contract holds.
    for n in 1..=PARTITION_WRITES {
        write_view(&cluster, LEADER, ((n - 1) % ITEMS) + 1, 1.0);
    }
    let leader_hwm = leader_seq(&cluster);
    println!("[step3] leader last_seq after partition-window writes = {leader_hwm}");

    // ap-south's applied STALLS at the pre-partition mark (the WAL ships never
    // arrive). Observe via the DIRECT addr (the proxy only severs PEER traffic).
    // Give the relay a moment to attempt + fail the ships, then assert no advance.
    std::thread::sleep(Duration::from_millis(500));
    let applied_now = applied(&cluster, AP_SOUTH);
    assert_eq!(
        applied_now, applied_at_partition,
        "ap-south applied must stall during the partition: was {applied_at_partition}, now {applied_now}"
    );
    assert!(
        leader_hwm > applied_now,
        "leader high-water-mark ({leader_hwm}) must exceed stalled ap-south applied ({applied_now}) — lag is real"
    );
    println!(
        "[step3] lag is real: leader hwm {leader_hwm} > ap-south applied {applied_now} (stalled)"
    );

    // The leader's aggregated /cluster/status reports ap-south unreachable with
    // worst-case lag (the HTTP status probe traverses the severed proxy).
    poll_until(
        Duration::from_secs(10),
        "leader status must show ap-south reachable:false",
        || {
            let body = cluster.get_json(LEADER, "/cluster/status");
            body["regions"]
                .as_array()
                .and_then(|rows| rows.iter().find(|r| r["name"].as_str() == Some("ap-south")))
                .is_some_and(|row| row["reachable"].as_bool() == Some(false))
        },
    );
    let status = cluster.get_json(LEADER, "/cluster/status");
    let ap_row = status["regions"]
        .as_array()
        .unwrap()
        .iter()
        .find(|r| r["name"].as_str() == Some("ap-south"))
        .unwrap();
    assert_eq!(ap_row["reachable"].as_bool(), Some(false));
    assert!(
        ap_row["lag_events"].as_u64().unwrap_or(0) >= 1,
        "unreachable ap-south must report worst-case lag: {ap_row}"
    );
    println!(
        "[step3] aggregated status: ap-south reachable:false lag_events={}",
        ap_row["lag_events"]
    );

    // The degraded scatter-gather: GET /sharded/feed on the leader → 200, degraded,
    // names ap-south unavailable, still returns live-shard items.
    let resp = cluster.get(
        LEADER,
        "/sharded/feed?profile=trending&limit=12&deadline_ms=1000",
    );
    assert_eq!(
        resp.status().as_u16(),
        200,
        "degraded sharded feed must still be a 200, never an error"
    );
    let feed: serde_json::Value = resp.json().unwrap();
    let sg = &feed["scatter_gather"];
    assert_eq!(
        sg["degraded"].as_bool(),
        Some(true),
        "sharded feed must be degraded with ap-south down: {feed}"
    );
    let unavailable: Vec<&str> = sg["unavailable_shards"]
        .as_array()
        .unwrap()
        .iter()
        .filter_map(|v| v.as_str())
        .collect();
    assert_eq!(
        unavailable,
        vec!["ap-south"],
        "exactly ap-south must be the unavailable shard: {sg}"
    );
    assert!(
        !feed["items"].as_array().unwrap().is_empty(),
        "degraded sharded feed must still return live-shard items: {feed}"
    );
    println!(
        "[step3] /sharded/feed degraded:true unavailable={unavailable:?} items={}",
        feed["items"].as_array().unwrap().len()
    );

    // ap-south's OWN /feed (direct addr) still serves its PRE-partition data —
    // eventual consistency, not an outage.
    let during_partition_feed = feed_pairs(&cluster, AP_SOUTH, "trending", ITEMS as u32);
    assert_eq!(
        during_partition_feed
            .iter()
            .map(|(id, _)| *id)
            .collect::<Vec<_>>(),
        pre_partition_feed
            .iter()
            .map(|(id, _)| *id)
            .collect::<Vec<_>>(),
        "partitioned ap-south must still serve its pre-partition item set"
    );
    println!(
        "[step3] ap-south direct /feed still serves {} pre-partition items (stale-read OK)",
        during_partition_feed.len()
    );
}

// ── UAT step 4: heal + reconcile, no loss, no duplication, < 100ms ──────────────

#[test]
fn mp_uat_step4_heal_reconcile_no_loss_no_dup() {
    let (rewrite, proxies) = proxied_rewrite(&["ap-south"]);
    let cluster = MultiProcCluster::start_with(ClusterOptions::new(3).with_rewrite(rewrite));

    const ITEMS: u64 = 12;
    /// The user whose hide we record during the window and verify converges.
    const HIDE_USER: u64 = 99;
    /// The item that user hides (one of the seeded, rankable items).
    const HIDE_ITEM: u64 = 3;

    seed_items_and_embeddings(&cluster, LEADER, ITEMS);
    for entity_id in 1..=ITEMS {
        write_view(&cluster, LEADER, entity_id, entity_id as f64);
    }
    cluster.wait_converged_all(convergence_budget());

    let ap = proxies.region("ap-south");

    // ── SEVER, then write + hide during the window ─────────────────────────────
    ap.sever_all();
    println!("[step4] severed ap-south from all peers");

    for n in 1..=PARTITION_WRITES {
        write_view(&cluster, LEADER, ((n - 1) % ITEMS) + 1, 1.0);
    }
    // Record a hard negative on the LEADER during the partition. Hides converge via
    // /cluster/reconcile (CRDT), NOT the WAL relay, so ap-south will not see it
    // until we reconcile after the heal.
    let resp = cluster.post(
        LEADER,
        "/hardnegs",
        &serde_json::json!({ "user_id": HIDE_USER, "item_id": HIDE_ITEM }),
    );
    assert_eq!(
        resp.status().as_u16(),
        204,
        "hardneg on leader during partition must 204: {}",
        resp.status()
    );
    let leader_hwm = leader_seq(&cluster);

    // ap-south is behind (writes did not ship). Confirm the real lag before heal.
    poll_until(
        Duration::from_secs(5),
        "ap-south applied should be behind the leader during the partition",
        || applied(&cluster, AP_SOUTH) < leader_hwm,
    );
    println!(
        "[step4] during partition: leader hwm={leader_hwm}, ap-south applied={}",
        applied(&cluster, AP_SOUTH)
    );

    // ── HEAL the link, then redeliver the missed segments ──────────────────────
    // Healing the link is necessary but not instantly sufficient: the leader's
    // gRPC circuit breaker opened during the partition and stays open for its reset
    // window, so the operator re-issues `/cluster/heal` until the breaker half-opens
    // and one probe ship closes it, redelivering every missed batch (real behavior).
    ap.heal_all();
    println!("[step4] healed ap-south link; driving redelivery through the breaker reset");

    heal_until_converged(&cluster, "ap-south", &[AP_SOUTH]);

    // No loss: ap-south applies up to the leader's high-water-mark.
    let ap_applied = applied(&cluster, AP_SOUTH);
    assert!(
        ap_applied >= leader_hwm,
        "after heal, ap-south applied ({ap_applied}) must reach leader hwm ({leader_hwm}) — zero loss"
    );
    println!("[step4] no loss: ap-south applied={ap_applied} >= leader hwm={leader_hwm}");

    // No duplication: leader↔ap-south feed-score parity to 1e-6 (a doubly-applied
    // segment would inflate decayed scores past the tolerance).
    let leader_feed = feed_pairs(&cluster, LEADER, "trending", ITEMS as u32);
    let ap_feed = feed_pairs(&cluster, AP_SOUTH, "trending", ITEMS as u32);
    assert_feed_parity("leader vs ap-south (post-heal)", &leader_feed, &ap_feed);
    println!(
        "[step4] no duplication: {} items match leader<->ap-south scores to 1e-6",
        leader_feed.len()
    );

    // Idempotent re-heal: redelivering again leaves the converged state unchanged
    // (the breaker is closed now, so this heal ships immediately; the receiver's
    // monotonic advance drops the re-shipped batches).
    let resp = cluster.post(
        LEADER,
        "/cluster/heal",
        &serde_json::json!({ "region": "ap-south" }),
    );
    assert_eq!(resp.status().as_u16(), 200, "second /cluster/heal must 200");
    cluster.wait_converged_all(convergence_budget());
    let ap_feed_again = feed_pairs(&cluster, AP_SOUTH, "trending", ITEMS as u32);
    assert_feed_parity("ap-south re-heal idempotent", &ap_feed, &ap_feed_again);
    println!("[step4] re-heal idempotent: ap-south feed unchanged");

    // ── RECONCILE the hard negative (CRDT path, not the WAL relay) ─────────────
    // Reconcile ON the leader WITH ap-south: the leader ships its snapshot (with
    // the hide) into ap-south's merge AND applies ap-south's snapshot back, so the
    // hide converges onto ap-south. Assert merge+apply < 100ms BOTH sides.
    let resp = cluster.post(
        LEADER,
        "/cluster/reconcile",
        &serde_json::json!({ "region": "ap-south" }),
    );
    assert_eq!(
        resp.status().as_u16(),
        200,
        "reconcile must 200: {}",
        resp.status()
    );
    let body: serde_json::Value = resp.json().unwrap();
    assert_eq!(body["ok"].as_bool(), Some(true));
    assert_eq!(body["region"].as_str(), Some("ap-south"));
    let local_ms = body["local_elapsed_ms"].as_u64().unwrap();
    let remote_ms = body["remote_elapsed_ms"].as_u64().unwrap();
    assert!(
        local_ms < 100 && remote_ms < 100,
        "reconcile merge+apply must be < 100ms both sides: local={local_ms}ms remote={remote_ms}ms"
    );
    println!("[step4] reconcile timings: local={local_ms}ms remote={remote_ms}ms (both < 100ms)");

    // "Hides remain hidden": the hidden item is FILTERED from ap-south's user-scoped
    // feed (the engine's user-context stage drops hard negatives when ?user_id is
    // set; verified by reading the engine retrieve path). A control read WITHOUT the
    // user still shows the item, proving the filter — not a disappearance.
    let global_ids = feed_pairs(&cluster, AP_SOUTH, "trending", ITEMS as u32)
        .into_iter()
        .map(|(id, _)| id)
        .collect::<Vec<_>>();
    assert!(
        global_ids.contains(&HIDE_ITEM),
        "control: item {HIDE_ITEM} must still rank in the un-scoped feed: {global_ids:?}"
    );
    let user_ids = feed_item_ids_for_user(&cluster, AP_SOUTH, HIDE_USER, "trending", ITEMS as u32);
    assert!(
        !user_ids.contains(&HIDE_ITEM),
        "hide must converge: item {HIDE_ITEM} must be ABSENT from user {HIDE_USER}'s feed on ap-south after reconcile: {user_ids:?}"
    );
    println!(
        "[step4] hide converged: item {HIDE_ITEM} present in global feed but hidden from user {HIDE_USER} on ap-south"
    );

    // Capture ap-south's exact feed scores immediately BEFORE the second reconcile,
    // so the comparison below isolates the reconcile's effect (not drift across the
    // hide-verification reads above).
    let ap_feed_pre_2nd = feed_pairs(&cluster, AP_SOUTH, "trending", ITEMS as u32);

    // Repeat reconcile → scores EXACTLY unchanged (bug 2 fixed). The CRDT signal merge
    // is now idempotent: `take_crdt_snapshot` attributes every node's contribution to
    // ONE canonical replication shard, so reconciling two already-converged nodes is a
    // fixpoint — the merged decay score does not move. Before the fix this crept
    // (0.5 → 0.375 → …) on every reconcile and was UNASSERTABLE; now we assert it.
    let resp = cluster.post(
        LEADER,
        "/cluster/reconcile",
        &serde_json::json!({ "region": "ap-south" }),
    );
    assert_eq!(resp.status().as_u16(), 200, "repeat reconcile must 200");
    let user_ids_after =
        feed_item_ids_for_user(&cluster, AP_SOUTH, HIDE_USER, "trending", ITEMS as u32);
    assert_eq!(
        user_ids, user_ids_after,
        "repeat reconcile must leave ap-south's user-scoped feed identical (idempotent)"
    );
    let ap_feed_final = feed_pairs(&cluster, AP_SOUTH, "trending", ITEMS as u32);
    // EXACT-unchanged on scores: the only divergence permitted is read-time decay
    // between the two feed reads (~microseconds → < 1e-9 relative at a 7-day
    // half-life), which is orders of magnitude below the ~25% bug-2 creep this guards.
    // assert_feed_parity's 1e-6 bound is far tighter than any creep and far looser than
    // the decay-between-reads noise, so it cleanly asserts "scores did not move".
    assert_feed_parity(
        "ap-south reconcile EXACT no-op on scores (bug 2 fixed)",
        &ap_feed_pre_2nd,
        &ap_feed_final,
    );
    // Also confirm stability against the original post-heal snapshot.
    assert_feed_parity("ap-south reconcile idempotent", &ap_feed, &ap_feed_final);
    println!(
        "[step4] repeat reconcile is an EXACT no-op on scores (bug 2 fixed): hide still effective, \
         feed unchanged"
    );
}

// ── ROADMAP criterion: partition between two followers ──────────────────────────

#[test]
fn mp_partition_between_followers() {
    // Proxy inbound edges to BOTH followers so we can sever a single directed edge.
    let (rewrite, proxies) = proxied_rewrite(&["eu-west", "ap-south"]);
    let cluster = MultiProcCluster::start_with(ClusterOptions::new(3).with_rewrite(rewrite));

    const ITEMS: u64 = 10;
    seed_items_and_embeddings(&cluster, LEADER, ITEMS);
    for entity_id in 1..=ITEMS {
        write_view(&cluster, LEADER, entity_id, entity_id as f64);
    }
    cluster.wait_converged_all(convergence_budget());

    // ── PHASE 1: sever ONLY the eu-west -> ap-south directed edge ──────────────
    // In a leader-ships topology NO replication traffic flows follower↔follower:
    // each follower's WAL segments arrive over its LEADER link. Severing the
    // eu-west↔ap-south edge is therefore INVISIBLE to correctness — both followers
    // keep converging via their intact leader links. This is the honest reading of
    // the ROADMAP "partition between two followers: writes continue on leader;
    // convergence preserved" criterion.
    proxies.edge("eu-west", "ap-south").sever_all();
    // Sever the reverse direction too, so the followers genuinely cannot reach each
    // other at all (a complete mutual follower partition), while both leader links
    // stay up.
    proxies.edge("ap-south", "eu-west").sever_all();
    println!("[followers] severed eu-west<->ap-south (both directions); leader links intact");

    // Writes continue on the leader and BOTH followers converge — proving the
    // follower↔follower partition does not impede the leader-ships fabric.
    for n in 1..=PARTITION_WRITES {
        write_view(&cluster, LEADER, ((n - 1) % ITEMS) + 1, 1.0);
    }
    cluster.wait_converged_all(convergence_budget());
    let leader_hwm = leader_seq(&cluster);
    let eu_applied = applied(&cluster, EU_WEST);
    let ap_applied = applied(&cluster, AP_SOUTH);
    assert!(
        eu_applied >= leader_hwm && ap_applied >= leader_hwm,
        "follower<->follower partition must NOT block convergence: leader hwm={leader_hwm}, eu={eu_applied}, ap={ap_applied}"
    );
    println!(
        "[followers] PROOF follower<->follower partition is invisible: leader hwm={leader_hwm}, eu applied={eu_applied}, ap applied={ap_applied} (both converged with the inter-follower link DOWN)"
    );

    // ── PHASE 2: ALSO sever the leader -> ap-south edge — NOW lag climbs ────────
    let leader_to_ap = proxies.edge("us-east", "ap-south");
    leader_to_ap.sever_all();
    println!("[followers] additionally severed us-east -> ap-south; lag should climb now");

    let pre_phase2_ap = applied(&cluster, AP_SOUTH);
    for n in 1..=PARTITION_WRITES {
        write_view(&cluster, LEADER, ((n - 1) % ITEMS) + 1, 1.0);
    }
    let leader_hwm2 = leader_seq(&cluster);
    // eu-west still converges (its leader link is up); ap-south stalls.
    poll_until(
        convergence_budget(),
        "eu-west must still converge with only ap-south's leader link severed",
        || applied(&cluster, EU_WEST) >= leader_hwm2,
    );
    std::thread::sleep(Duration::from_millis(500));
    let ap_during = applied(&cluster, AP_SOUTH);
    assert_eq!(
        ap_during, pre_phase2_ap,
        "with leader->ap-south severed, ap-south applied must stall: was {pre_phase2_ap}, now {ap_during}"
    );
    assert!(
        leader_hwm2 > ap_during,
        "contrast: leader hwm ({leader_hwm2}) now exceeds stalled ap-south applied ({ap_during})"
    );
    println!(
        "[followers] contrast confirmed: leader hwm={leader_hwm2} > ap-south applied={ap_during} (stalled) while eu-west converged"
    );

    // ── HEAL: convergence restored with no data loss ───────────────────────────
    // The leader->ap-south ships tripped the breaker, so drive the heal through the
    // breaker reset exactly as the operator would.
    leader_to_ap.heal_all();
    heal_until_converged(&cluster, "ap-south", &[AP_SOUTH]);
    let ap_final = applied(&cluster, AP_SOUTH);
    assert!(
        ap_final >= leader_hwm2,
        "after heal, ap-south applied ({ap_final}) must reach leader hwm ({leader_hwm2}) — no data loss"
    );
    let leader_feed = feed_pairs(&cluster, LEADER, "trending", ITEMS as u32);
    let ap_feed = feed_pairs(&cluster, AP_SOUTH, "trending", ITEMS as u32);
    assert_feed_parity(
        "leader vs ap-south (followers test heal)",
        &leader_feed,
        &ap_feed,
    );
    println!(
        "[followers] healed: ap-south applied={ap_final} >= leader hwm={leader_hwm2}, feed parity 1e-6"
    );
}

/// m11p2 exit-gate test — ONE replicated log, end to end with real processes:
///
/// 1. **Items written anywhere are readable everywhere**: items + embeddings
///    seeded on the leader (and one item written THROUGH a follower, which
///    forwards) reach every node via kind-1/2 WAL records on the same stream
///    as signals — there is no HTTP broadcast left to deliver them.
/// 2. **Downtime heals by pure log catch-up**: a follower stopped while the
///    leader keeps writing items/embeddings/signals restarts and converges
///    via its boot-time `StreamSegments` pull over the leader's durable WAL —
///    no `/cluster/heal`, no O(items) HTTP traffic, no operator verb at all.
#[test]
fn mp_items_ride_the_log_and_catchup_stream() {
    let mut cluster = MultiProcCluster::start(3);

    const ITEMS: u64 = 6;
    seed_items_and_embeddings(&cluster, LEADER, ITEMS);
    for entity_id in 1..=ITEMS {
        write_view(&cluster, LEADER, entity_id, entity_id as f64);
    }
    // One item written THROUGH a follower gateway: it forwards to the leader,
    // rides the same log, and must become readable everywhere.
    let via_follower = ITEMS + 1;
    let resp = cluster.post(
        EU_WEST,
        "/items",
        &serde_json::json!({
            "entity_id": via_follower,
            "metadata": { "title": "item via follower" }
        }),
    );
    assert_eq!(
        resp.status().as_u16(),
        201,
        "follower /items must forward to the leader and 201"
    );
    write_view(&cluster, LEADER, via_follower, 1.0);

    cluster.wait_converged_all(convergence_budget());
    let leader_feed = feed_pairs(&cluster, LEADER, "trending", (via_follower + 2) as u32);
    assert!(
        leader_feed.iter().any(|(id, _)| *id == via_follower),
        "the follower-written item must rank on the leader: {leader_feed:?}"
    );
    for follower in [EU_WEST, AP_SOUTH] {
        let feed = feed_pairs(&cluster, follower, "trending", (via_follower + 2) as u32);
        assert_feed_parity("leader vs follower (items via log)", &leader_feed, &feed);
    }
    println!(
        "[log] {via_follower} items (one via a follower gateway) + signals converged on \
         every node through the one replicated log"
    );

    // ── Downtime → restart → boot-time catch-up stream (no heal verb) ─────────
    cluster.stop_graceful(AP_SOUTH);
    let offline_first = via_follower + 1;
    let offline_last = via_follower + 4;
    for entity_id in offline_first..=offline_last {
        let resp = cluster.post(
            LEADER,
            "/items",
            &serde_json::json!({
                "entity_id": entity_id,
                "metadata": { "title": format!("offline item {entity_id}") }
            }),
        );
        assert_eq!(resp.status().as_u16(), 201, "leader /items during downtime");
        #[allow(clippy::cast_precision_loss)]
        let v = entity_id as f32;
        let resp = cluster.post(
            LEADER,
            "/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 during downtime"
        );
        write_view(&cluster, LEADER, entity_id, 2.0);
    }

    // Restart the follower: its boot-time catch-up request pulls the missed
    // range from the leader's durable WAL over StreamSegments. NO heal verb,
    // NO HTTP item traffic — convergence must be fully self-driving.
    cluster.restart(AP_SOUTH, &[]);
    cluster.wait_healthy(AP_SOUTH);
    cluster.wait_converged_all(convergence_budget());

    let leader_feed = feed_pairs(&cluster, LEADER, "trending", (offline_last + 2) as u32);
    let ap_feed = feed_pairs(&cluster, AP_SOUTH, "trending", (offline_last + 2) as u32);
    for entity_id in offline_first..=offline_last {
        assert!(
            ap_feed.iter().any(|(id, _)| *id == entity_id),
            "restarted follower must rank offline-written item {entity_id} \
             (delivered by the catch-up stream): {ap_feed:?}"
        );
    }
    assert_feed_parity(
        "leader vs restarted ap-south (catch-up stream)",
        &leader_feed,
        &ap_feed,
    );
    println!(
        "[log] restarted follower converged via the boot-time StreamSegments pull: \
         items {offline_first}..={offline_last} present with feed parity 1e-6"
    );
}