//! m11p5 §2.2–§2.7 — the RESEED path over real OS processes (tier 3, B3).
//!
//! Two tier-3 proofs that a behind-or-divergent node self-heals through a
//! boot-time snapshot install + catch-up stream, with ZERO operator verbs other
//! than process restarts (no `wipe_data_dir`, no manual data-dir surgery):
//!
//! 1. `mp_follower_reseeds_via_snapshot_after_compaction` — exit-gate 1's
//!    *mechanics* minus seed-join. A follower stops; the leader writes a batch
//!    big enough to ROTATE its WAL past one segment, then gracefully restarts so
//!    its clean shutdown COMPACTS the WAL below the stopped follower's resume
//!    seq. The follower restarts: its boot-time catch-up pull hits the leader's
//!    typed `snapshot-required` refusal (the segments below its frontier were
//!    compacted away), durably latches `reseed_required`, and keeps serving
//!    degraded. A second restart performs the marker boot — discover the leader,
//!    adopt its term, fetch the snapshot, swap it in — then converges via the
//!    catch-up stream. Every written item is then readable on the reseeded
//!    follower and its status no longer reports `reseed_required`.
//!
//!    Auto-election is pinned OFF (`auto_election: false`): the reseed + the
//!    legacy `/cluster/promote` re-leadership both work in the term-0 topology
//!    era, and a manual drill must not race the m11p4 failure detector.
//!
//! 2. `mp_quarantined_node_reseeds_without_wipe` — exit-gate 3. Under
//!    `ack=leader` load (a divergent suffix the cluster never quorum-acked), the
//!    leader is `SIGKILL`ed; the survivors elect; the old leader restarts and
//!    QUARANTINES on its divergent suffix — which now latches the reseed marker
//!    (`quarantined: true` AND `reseed_required: true`). Restarting it once more
//!    performs the marker boot: it snapshot-installs from the CURRENT ELECTED
//!    leader (the discovery loop + term-adoption path), converges, serves reads,
//!    and reports `quarantined: false` + `reseed_required: false` with the
//!    process-local divergence gauge cleared — and NO `wipe_data_dir` anywhere.
//!
//! # Budget
//!
//! Tier-3 over real OS processes. Two tests; each boots one fresh 3-process
//! cluster (the binary build is amortized to a `cargo build` no-op), writes a
//! bounded batch, and performs a bounded number of restarts + convergence waits.
//! Every wait is poll-with-deadline (no bare sleeps as correctness gates).
//! Whole-suite wall budget < 3 minutes on a developer laptop; each test's waits
//! sum well under 90s.
//!
//! ```bash
//! cargo test -p tidal-server --features cluster-e2e --test cluster_reseed -- --nocapture
//! ```
#![cfg(feature = "cluster-e2e")]
// Tier-3 harness allows, mirroring the sibling suites: `unwrap` on known-good
// fixtures is idiomatic test noise; the lossy numeric casts are the same
// pervasive-and-intentional scoring/sizing 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
)]

mod support;

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

use support::multiproc::{ClusterOptions, MultiProcCluster, convergence_budget};

/// Region 0 is the initial topology leader.
const LEADER: usize = 0;
/// Region 1 (`eu-west`) stays up throughout test 1 (a continuously-live voter).
const EU_WEST: usize = 1;
/// Region 2 (`ap-south`) is the follower that stops, falls behind a compacted
/// leader, and reseeds.
const AP_SOUTH: usize = 2;

/// Operator-driven determinism: pin auto-election OFF so the legacy
/// `/cluster/promote` verb (a term-0 topology-era fan-out) works and the m11p4
/// failure detector cannot race the manual drill. The reseed itself is the same
/// `FetchSnapshot` path in both eras — test 2 exercises the elected era.
const LEGACY_ELECTION_YAML: &str = "election:\n  auto_election: false";

/// The fast election block test 2 runs with (the elected-era reseed). Constraint
/// (the C2 bound enforced at topology load): `lease (350) + heartbeat (100) <
/// timeout_min (500)`.
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";

/// The PRODUCTION election timers (verbatim from `k8s/cluster/topology-configmap.yaml`):
/// heartbeat 300ms, election timeout 1500–3000ms, lease 900ms. The graceful-
/// rolling-restart repro runs with THESE, not the much faster `FAST_ELECTION_YAML`:
/// the production blocker is what a real k8s RollingUpdate does to the real
/// deployment, and the 3×-tighter fast timers manufacture extra mid-restart
/// re-election churn that the live cluster never sees. With production timers a
/// follower restart does not move leadership and a leader restart moves it exactly
/// once — the faithful scenario.
const PROD_ELECTION_YAML: &str = "election:\n  heartbeat_interval_ms: 300\n  election_timeout_min_ms: 1500\n  election_timeout_max_ms: 3000\n  leader_lease_ms: 900";

/// Items the leader writes while the follower is offline. Enough to MATTER
/// (well past a handful) and — paired with the per-item metadata blob (see
/// [`blob_value`]) — to push the WAL past one 16 MiB segment so the
/// graceful-shutdown compaction actually deletes the segments below the
/// follower's frontier (a single segment is never deleted; the snapshot path
/// only triggers once history is genuinely compacted away).
///
/// Sizing (m12 rolling-restart tuning): the graceful-shutdown compaction now
/// RETAINS the `WAL_RETENTION_SEGMENTS` (= 16) most-recent sealed segments so a
/// briefly-down follower can stream-catch-up instead of reseeding. For this
/// exit-gate to STILL force the snapshot-required path, the offline batch must
/// compact the follower's frontier segment away DESPITE retention — i.e. produce
/// more than `WAL_RETENTION_SEGMENTS + 1` = 17 segments. At ~56 KiB/item the
/// 16 MiB segment holds ~292 items, so 5600 items ≈ 19 segments ⇒ the follower's
/// frontier segment is well past the 16-segment retention window and is genuinely
/// deleted. (Sized off the constant + a 3-segment margin so a future `N` bump
/// never silently un-sizes this gate.)
const OFFLINE_ITEMS: u64 = 5600;

/// The engine caps item metadata at 8 KiB per VALUE and 64 KiB TOTAL per item
/// (a hard query-index-integrity invariant — never to be weakened). So the WAL
/// inflation rides MULTIPLE non-indexed values just under those caps: 8 keys ×
/// ~7 KiB ≈ 56 KiB total per item, comfortably below the 64 KiB total cap.
const BLOB_KEYS: usize = 8;
/// Bytes per blob value (< the 8 KiB per-value cap).
const BLOB_VALUE_BYTES: usize = 7 * 1024;

/// Build the per-item non-indexed metadata blob: [`BLOB_KEYS`] keys, each a
/// [`BLOB_VALUE_BYTES`]-byte filler value. Only `title` is a schema text field,
/// so these `blobN` keys ride the WAL (kind-1 item blob) without any Tantivy
/// indexing cost.
fn blob_value() -> String {
    "z".repeat(BLOB_VALUE_BYTES)
}

/// A handshake window LOW enough that the marker-boot install completes (or
/// falls back) inside the test budget rather than the 60 s production default.
/// 30 s is comfortably above a loopback discover+fetch yet bounded.
const RESEED_HANDSHAKE_MS: &str = "30000";

/// The harness's fixed region roster (mirrors `multiproc::region_name`): the
/// `proxied_rewrite` set and the proxy-edge lookups key on these names, and they
/// are stable across the cluster's lifetime.
fn cluster_region_name(idx: usize) -> String {
    const ROSTER: [&str; 3] = ["us-east", "eu-west", "ap-south"];
    ROSTER[idx].to_string()
}

/// A unique all-alpha search token for an entity id (digits 0-9 → letters a-j),
/// so `/search?query=<token>` is an exact item-presence probe (mirrors the
/// m11p3/m11p4 ledger checkers).
fn item_token(entity_id: u64) -> String {
    let mut token = String::from("rsd");
    for d in entity_id.to_string().bytes() {
        token.push(char::from(b'a' + (d - b'0')));
    }
    token
}

/// Write one item (small searchable `title` token + several large non-indexed
/// `blobN` values that inflate the WAL within the engine's metadata caps) plus
/// its 4-dim embedding and a `view` signal to the node at `idx`, asserting the
/// leader-durable contract (201 / 204 / 204). `blob` is the shared filler value
/// (built once by the caller); `heavy` toggles the WAL-inflating blob keys.
fn write_heavy_item(
    cluster: &MultiProcCluster,
    idx: usize,
    entity_id: u64,
    blob: &str,
    heavy: bool,
) {
    let mut metadata = serde_json::Map::new();
    metadata.insert("title".into(), item_token(entity_id).into());
    if heavy {
        for k in 0..BLOB_KEYS {
            metadata.insert(format!("blob{k}"), blob.into());
        }
    }
    let resp = cluster.post(
        idx,
        "/items",
        &serde_json::json!({ "entity_id": entity_id, "metadata": metadata }),
    );
    assert_eq!(resp.status().as_u16(), 201, "leader /items must 201");

    let v = entity_id as f32;
    let resp = cluster.post(
        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");

    let resp = cluster.post(
        idx,
        "/signals",
        &serde_json::json!({ "entity_id": entity_id, "signal": "view", "weight": 1.0 }),
    );
    assert_eq!(resp.status().as_u16(), 204, "leader /signals must 204");
}

/// [`write_heavy_item`] with a bounded retry, for the multi-group fixture.
///
/// Which survivor inherits a stopped node's groups varies per run, so a write
/// issued at any one gateway may be local for one group and a cross-group FORWARD
/// for another, and a forward issued inside an election window legitimately answers
/// a retryable 503. Retrying keeps the FIXTURE deterministic without masking a hard
/// failure: after the budget it still fails, and it reports the last status.
fn write_heavy_item_retrying(
    cluster: &MultiProcCluster,
    idx: usize,
    entity_id: u64,
    blob: &str,
    heavy: bool,
) {
    let deadline = Instant::now() + Duration::from_secs(30);
    loop {
        let mut metadata = serde_json::Map::new();
        metadata.insert("title".into(), item_token(entity_id).into());
        if heavy {
            for k in 0..BLOB_KEYS {
                metadata.insert(format!("blob{k}"), blob.into());
            }
        }
        let status = cluster
            .post(
                idx,
                "/items",
                &serde_json::json!({ "entity_id": entity_id, "metadata": metadata }),
            )
            .status()
            .as_u16();
        if status == 201 {
            // Embedding + signal are best-effort here: the gate is WAL volume on the
            // groups, which the item blob already provides.
            let v = entity_id as f32;
            let _ = cluster.post(
                idx,
                "/embeddings",
                &serde_json::json!({ "entity_id": entity_id, "values": [v, v + 1.0, v + 2.0, v + 3.0] }),
            );
            return;
        }
        assert!(
            Instant::now() < deadline,
            "item {entity_id} never accepted (last status {status}); a cross-group forward is \
             failing beyond the election window, which is a different defect than this gate covers"
        );
        std::thread::sleep(Duration::from_millis(250));
    }
}

/// Whether `/search?query=<token>` on node `idx` returns `entity_id` (an exact
/// item-presence probe — the content invariant).
fn item_searchable(cluster: &MultiProcCluster, idx: usize, entity_id: u64) -> bool {
    let token = item_token(entity_id);
    let found = cluster.get_json(idx, &format!("/search?query={token}&limit=5"));
    found["items"]
        .as_array()
        .is_some_and(|r| r.iter().any(|x| x["entity_id"].as_u64() == Some(entity_id)))
}

/// The `first_seq` of every WAL segment retained on node `idx`, ascending.
///
/// Segment filenames are `wal-{first_seq:020}.seg` for the single-shard layout
/// (`tidal::wal::segment::segment_filename`), so the retained history floor is
/// the FIRST entry: the source can serve a catch-up pull from that seq onward and
/// no earlier.
fn retained_segment_first_seqs(cluster: &MultiProcCluster, idx: usize) -> Vec<u64> {
    let wal_dir = cluster.data_dir(idx).join("wal");
    let mut seqs: Vec<u64> = std::fs::read_dir(&wal_dir)
        .into_iter()
        .flatten()
        .flatten()
        .filter_map(|e| {
            let name = e.file_name().to_string_lossy().into_owned();
            let rest = name.strip_prefix("wal-")?.strip_suffix(".seg")?;
            rest.rsplit('-').next()?.parse::<u64>().ok()
        })
        .collect();
    seqs.sort_unstable();
    seqs
}

/// Assert the PREMISE of the compaction gate: after the leader's graceful
/// restart, its retained WAL must no longer cover `resume_seq`.
///
/// This exists because the gate previously asserted only its CONSEQUENCE
/// (`reseed_required == true`). When the premise silently stopped holding, the
/// test failed 40s later pointing at the follower — which looked like a reseed
/// bug and was actually "the leader never compacted anything". Assert the setup
/// so an un-sized fixture names itself.
fn assert_history_compacted_past(cluster: &MultiProcCluster, idx: usize, resume_seq: u64) {
    let seqs = retained_segment_first_seqs(cluster, idx);
    let floor = seqs.first().copied().unwrap_or(0);
    assert!(
        !seqs.is_empty(),
        "leader has no WAL segments at all; the fixture is broken, not the reseed path"
    );
    assert!(
        floor > resume_seq,
        "PREMISE FAILED: the leader still serves the follower's resume seq, so no \
         snapshot-required refusal can ever happen and this gate proves nothing. \
         retained segment first_seqs = {seqs:?} (floor {floor}) must all exceed \
         resume_seq {resume_seq}. Either the offline batch no longer rotates past \
         WAL_RETENTION_SEGMENTS (re-size OFFLINE_ITEMS against that constant and the \
         16 MiB segment size) or the graceful-shutdown compaction stopped deleting."
    );
}

/// Read a status field as a bool (defaulting `false` when the node is
/// unreachable or the field is absent).
fn status_bool(cluster: &MultiProcCluster, idx: usize, field: &str) -> bool {
    cluster
        .local_status(idx)
        .and_then(|s| s[field].as_bool())
        .unwrap_or(false)
}

/// Poll node `idx`'s own `/cluster/status/local` until `field == expected`, or
/// the deadline. Returns on success; panics with the last seen status on
/// timeout.
fn await_status_bool(
    cluster: &MultiProcCluster,
    idx: usize,
    field: &str,
    expected: bool,
    budget: Duration,
    what: &str,
) {
    let deadline = Instant::now() + budget;
    loop {
        if status_bool(cluster, idx, field) == expected {
            return;
        }
        assert!(
            Instant::now() < deadline,
            "{what}: node {idx} did not report {field}={expected} within {budget:?}; \
             last status: {:?}",
            cluster.local_status(idx)
        );
        std::thread::sleep(Duration::from_millis(100));
    }
}

/// Poll until node `idx`'s OWN status reports zero lag and an applied frontier
/// at or above the live leader's `last_seq` (convergence proven against the
/// node's own process boundary, never the leader's aggregate view).
fn await_self_converged(cluster: &MultiProcCluster, idx: usize, budget: Duration, what: &str) {
    let deadline = Instant::now() + budget;
    loop {
        let leader_seq = cluster.leader_last_seq();
        if let (Some(target), Some(st)) = (leader_seq, cluster.local_status(idx)) {
            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 {
                return;
            }
        }
        assert!(
            Instant::now() < deadline,
            "{what}: node {idx} did not converge (lag 0 vs leader last_seq) within {budget:?}; \
             leader_seq={leader_seq:?} status={:?}",
            cluster.local_status(idx)
        );
        std::thread::sleep(Duration::from_millis(100));
    }
}

/// Promote `region` to leader via the legacy `/cluster/promote` verb, asserting
/// the 200, then wait for every LIVE node to agree on the leader. Used in the
/// legacy (term-0) drill to restore a leader after the original leader's
/// graceful restart booted it as a follower (the durable-state §1.4-1 rule).
fn promote_and_agree(cluster: &MultiProcCluster, via_idx: usize, region: &str) {
    let resp = cluster.post(
        via_idx,
        "/cluster/promote",
        &serde_json::json!({ "region": region }),
    );
    assert_eq!(
        resp.status().as_u16(),
        200,
        "/cluster/promote must 200: {}",
        resp.status()
    );
    cluster.wait_leader_agreed(region, convergence_budget());
}

/// Write ONLY the heavy item (no embedding / no signal) to node `idx`, asserting
/// the 201. Used by the graceful-rolling-restart repro where the only thing that
/// matters is WAL volume (the 56 KiB item blob fills segments to force
/// compaction) and item-presence read-back — so the two extra POSTs per item the
/// full [`write_heavy_item`] issues are pure cost. Cutting them ~3×'s the seed
/// throughput and drops the embedding/signal memory the full writer would hold.
fn write_heavy_item_only(cluster: &MultiProcCluster, idx: usize, entity_id: u64, blob: &str) {
    let mut metadata = serde_json::Map::new();
    metadata.insert("title".into(), item_token(entity_id).into());
    for k in 0..BLOB_KEYS {
        metadata.insert(format!("blob{k}"), blob.into());
    }
    let resp = cluster.post(
        idx,
        "/items",
        &serde_json::json!({ "entity_id": entity_id, "metadata": metadata }),
    );
    assert_eq!(resp.status().as_u16(), 201, "leader /items must 201");
}

/// The index of the current live leader (by `is_leader`), or `None` if no live
/// node reports leadership yet. Mirrors `cluster_membership::current_leader_idx`
/// — the elected-era analogue of the fixed `LEADER` constant.
fn current_leader_idx(cluster: &MultiProcCluster) -> Option<usize> {
    (0..cluster.len()).find(|&i| {
        cluster
            .local_status(i)
            .is_some_and(|s| s["is_leader"].as_bool() == Some(true))
    })
}

/// Wait until every live node has caught up to within `max_lag` of the leader —
/// the test-side analogue of a k8s rolling update gating each pod restart on the
/// previous pod's READINESS. Without it the test fires restarts back-to-back, so a
/// leader's step-down drain (which needs a quorum of CAUGHT-UP followers to commit
/// its tail) can stall behind a follower still catching up from its own restart,
/// and that leader then diverges. A real rollout never restarts the next pod until
/// the last one is ready; this reproduces that pacing. Tolerant of a small in-flight
/// tail under sustained load (hence `max_lag`, not strict 0).
fn await_cluster_caught_up(cluster: &MultiProcCluster, max_lag: u64, budget: Duration, what: &str) {
    let deadline = Instant::now() + budget;
    loop {
        let all = (0..cluster.len()).all(|i| {
            cluster
                .local_status(i)
                .and_then(|s| s["lag_events"].as_u64())
                .is_some_and(|lag| lag <= max_lag)
        });
        if all {
            return;
        }
        assert!(
            Instant::now() < deadline,
            "{what}: cluster did not catch up (all nodes lag <= {max_lag}) within {budget:?}; \
             last: {:?}",
            (0..cluster.len())
                .map(|i| cluster
                    .local_status(i)
                    .and_then(|s| s["lag_events"].as_u64()))
                .collect::<Vec<_>>()
        );
        std::thread::sleep(Duration::from_millis(100));
    }
}

/// Force the cluster out of the term-0 TOPOLOGY era into the ELECTED era before a
/// measured rolling restart. A freshly-booted cluster can hold leadership at term 0
/// (the topology leader leads without an election ever firing), but PRODUCTION is
/// always elected-era — a long-running cluster has elected many terms by the time
/// it is rolled. Restarting the current leader once forces the survivors to elect
/// (term ≥ 1); we then wait until a node reports an elected term and the cluster
/// re-agrees a leader, so the subsequent measured restarts exercise only the
/// elected-era step-down path (where the graceful leadership hand-off applies).
fn ensure_elected_era(cluster: &mut MultiProcCluster) {
    let leader = current_leader_idx(cluster).expect("a leader before forcing elected era");
    cluster.restart_graceful(leader, &[]);
    let _ = cluster.wait_shard_leaders_agreed(convergence_budget() + Duration::from_secs(20));
    let deadline = Instant::now() + convergence_budget();
    loop {
        let elected = (0..cluster.len()).any(|i| {
            cluster
                .local_status(i)
                .and_then(|s| s["term"].as_u64())
                .is_some_and(|t| t >= 1)
        });
        if elected {
            return;
        }
        assert!(
            Instant::now() < deadline,
            "cluster did not reach the elected era (term >= 1) within {:?}",
            convergence_budget()
        );
        std::thread::sleep(Duration::from_millis(100));
    }
}

/// Gate-1 mechanics (minus seed-join): a follower that fell behind a leader
/// whose WAL has been COMPACTED past its resume seq reseeds via the boot-time
/// snapshot install — zero operator verbs other than restarts.
#[test]
fn mp_follower_reseeds_via_snapshot_after_compaction() {
    // Pin auto-election OFF: legacy promote + reseed both ride the term-0 era.
    // Give the AP_SOUTH node the LOW handshake window so its marker boot does
    // not wait the 60 s production default before installing.
    let opts = ClusterOptions::new(3)
        .with_topology_extra(LEGACY_ELECTION_YAML)
        .with_env(AP_SOUTH, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS);
    let mut cluster = MultiProcCluster::start_with(opts);

    // A couple of baseline items, fully converged, so every node shares a
    // common prefix before the follower goes offline.
    for entity in 1..=3u64 {
        write_heavy_item(&cluster, LEADER, entity, "", false);
    }
    cluster.wait_converged_all(convergence_budget());
    println!("[reseed] baseline converged on all three nodes");

    // ── Stop one follower gracefully: its applied frontier latches here. ──────
    cluster.stop_graceful(AP_SOUTH);
    let frontier_at_stop = cluster
        .local_status(LEADER)
        .and_then(|s| s["last_seq"].as_u64())
        .unwrap_or(0);
    println!("[reseed] ap-south stopped; leader frontier at stop = {frontier_at_stop}");

    // ── The leader writes a batch big enough to ROTATE its WAL past one
    // segment (large non-indexed blobs), so the graceful-restart compaction
    // below actually deletes the segments holding the follower's resume range.
    let blob = blob_value();
    for entity in 4..=(3 + OFFLINE_ITEMS) {
        write_heavy_item(&cluster, LEADER, entity, &blob, true);
    }
    // EU_WEST (continuously live) must converge the batch, so a leader exists to
    // reseed from after the original leader restarts.
    await_self_converged(
        &cluster,
        EU_WEST,
        convergence_budget(),
        "eu-west catches the offline batch",
    );
    println!("[reseed] leader wrote {OFFLINE_ITEMS} heavy items; eu-west converged");

    // ── GRACEFULLY restart the LEADER: clean shutdown writes the WAL checkpoint
    // marker and COMPACTS the WAL — the stopped follower's resume seq is now
    // below the leader's earliest servable seq. The restarted leader boots as a
    // FOLLOWER (durable §1.4-1 rule), so re-promote it (legacy term-0 verb) to
    // restore a leader for the follower to reseed from.
    cluster.restart_graceful(LEADER, &[]);
    promote_and_agree(&cluster, LEADER, cluster.region_name(LEADER));
    println!("[reseed] leader gracefully restarted (WAL compacted) and re-promoted");

    // PREMISE CHECK (see `assert_history_compacted_past`): the follower resumes at
    // `frontier_at_stop + 1`, so the leader's retained WAL must no longer cover it.
    // Without this, an un-sized fixture silently turns the gate into a no-op that
    // fails 40s later looking like a follower bug.
    assert_history_compacted_past(&cluster, LEADER, frontier_at_stop + 1);
    println!(
        "[reseed] premise holds: leader retains {:?}, follower resume seq {} is gone",
        retained_segment_first_seqs(&cluster, LEADER),
        frontier_at_stop + 1
    );

    // ── Restart the stopped follower (first boot: NO marker yet). Its runtime
    // boot-time catch-up pull requests from its frontier+1, which the leader's
    // compacted WAL cannot serve → typed `snapshot-required` refusal → the
    // durable marker latches. The node keeps serving degraded.
    cluster.restart(AP_SOUTH, &[]);
    await_status_bool(
        &cluster,
        AP_SOUTH,
        "reseed_required",
        true,
        convergence_budget() + Duration::from_secs(10),
        "the boot-time pull against a compacted leader latches reseed_required",
    );
    println!("[reseed] ap-south restarted behind a compacted leader → reseed_required latched");

    // ── Restart the follower ONCE MORE: the marker boot performs the snapshot
    // install (discover leader → adopt term → fetch → swap), then converges via
    // the catch-up stream. `reseed_self_restart` defaults to false, so the test
    // restarts it explicitly (the LOW handshake window keeps the install bounded).
    cluster.restart(AP_SOUTH, &[]);
    await_self_converged(
        &cluster,
        AP_SOUTH,
        convergence_budget() + Duration::from_secs(20),
        "the marker boot installs the snapshot and the follower converges",
    );
    println!("[reseed] ap-south marker boot installed the snapshot and converged (lag 0)");

    // ── CONTENT: every offline-written item is readable on the reseeded
    // follower (the snapshot carried the compacted history; the stream carried
    // the suffix). Poll past the ~2 s text-index auto-commit.
    // Convergence-class wait: the snapshot install + text-index rebuild +
    // ~2 s auto-commit compete with suite-parallel load, so a fixed short
    // budget flakes on busy hosts; use the harness convergence budget
    // (30 s default, TIDAL_TEST_CONVERGENCE_BUDGET_SECS to widen).
    let probe_deadline = Instant::now() + convergence_budget();
    for entity in [4u64, 50, 120, 200, 3 + OFFLINE_ITEMS] {
        loop {
            if item_searchable(&cluster, AP_SOUTH, entity) {
                break;
            }
            assert!(
                Instant::now() < probe_deadline,
                "reseeded follower is missing item {entity} (snapshot+stream must \
                 deliver every written item)"
            );
            std::thread::sleep(Duration::from_millis(200));
        }
    }
    println!("[reseed] every probed offline item is searchable on the reseeded follower");

    // ── The marker is gone: the swap discarded the old data dir (with the
    // marker) and the new dir from staging carries none.
    assert!(
        !status_bool(&cluster, AP_SOUTH, "reseed_required"),
        "the reseeded follower must no longer report reseed_required: {:?}",
        cluster.local_status(AP_SOUTH)
    );
    assert!(
        !status_bool(&cluster, AP_SOUTH, "quarantined"),
        "the reseeded follower must not be quarantined: {:?}",
        cluster.local_status(AP_SOUTH)
    );
    println!("[reseed] ap-south reseeded clean: reseed_required=false, quarantined=false");
}

/// The PRODUCTION shape (2026-08-20 incident): a node that hosts SEVERAL shard
/// groups and must reseed MORE THAN ONE of them must converge — not heal one
/// group per process restart forever.
///
/// Live tidaldb-0 restarted 8 times in 22 minutes on the served-evidence fix with
/// the exiting group alternating 2 -> 1 -> 2. Each restart streamed a fresh
/// snapshot of a 33k x 1536-dim corpus off the healthy leaders, so the loop is
/// expensive as well as non-terminating. Nothing in the suite covered a
/// multi-group reseed: `mp_follower_reseeds_via_snapshot_after_compaction` is
/// single-group, so a per-group `reseed_self_restart` that never reaches a
/// fixpoint was invisible.
///
/// Shape, matching `k8s/cluster/topology-configmap.yaml`: 3 nodes x 3 groups,
/// full placement, balanced term-0 leaders (group `s` led by node `s`), and the
/// PRODUCTION election timers. Node 2 goes down, so its group-2 leadership moves
/// and one surviving node ends up leading two groups — the live arrangement.
///
/// The gate is a FIXPOINT: node 2 must end Ready, reporting no reseed marker, with
/// content readable from every group, within a bounded number of restarts.
///
/// # This gate CLOSED the multi-group reseed defect
///
/// It was written as a reproduction and failed exactly as production did — the node
/// announced itself settled and was missing data:
///
/// ```text
/// [multi] node 2 settled after 0 orchestrator restart(s)
/// [multi] node 2 exited AFTER settling; orchestrator reboot #1
/// missing item 500 (reboots=1) ... reseed_required: false, lag_events: 0,
///                                   applied_events: 3798
/// ```
///
/// Root cause was readiness asserting on ABSENCE of bad news. `is_ready` gated
/// convergence behind `install_boot || seed_joiner`, so a plain restarted voter was
/// Ready on arrival — before it had learned the leader's frontier — and
/// `lag_events` could not contradict it, being `leader_seqno - applied` on a gauge
/// that reads 0 until the frontier is known (`0 - 0 = 0`). A blind node therefore
/// looked converged. Convergence is now required for EVERY boot against a KNOWN
/// frontier, and only ESTABLISHED (election-runtime) leadership self-certifies —
/// never the topology's term-0 belief, which would let a booting node certify the
/// group it merely thinks it leads.
///
/// Post-fix run, all three groups converged against real frontiers:
///
/// ```text
/// shard 0: applied 3797, leader_seqno 3797, term 1
/// shard 1: applied 3726, leader_seqno 3726, term 5
/// shard 2: applied 3747, leader_seqno 3747, term 3
/// [multi] every probed item is readable ... (2 reboot(s) total)
/// ```
///
/// Runs long (~15 min) and wants widened tier-3 budgets:
///
/// ```bash
/// TIDAL_TEST_BOOT_BUDGET_SECS=300 TIDAL_TEST_CONVERGENCE_BUDGET_SECS=180 \
///   cargo test -p tidal-server --features cluster-e2e --test cluster_reseed \
///   mp_multi_group_node_converges_after_reseeding_several_groups -- --nocapture
/// ```
#[test]
fn mp_multi_group_node_converges_after_reseeding_several_groups() {
    const GROUPS: usize = 3;
    /// Generous but FINITE. One restart per group that needs a reseed is the
    /// designed cost (the exit is process-wide); anything beyond that is the
    /// non-terminating loop this gate exists to catch.
    const RESTART_CEILING: u32 = GROUPS as u32 + 2;

    // `reseed_self_restart: true` is what PRODUCTION sets
    // (k8s/cluster/topology-configmap.yaml) and what makes the loop possible: a
    // latched marker drains and exits(0) so the orchestrator's next boot installs.
    // The harness default is false, which is precisely why no existing test could
    // ever observe a non-terminating multi-group reseed.
    let topology = format!("{PROD_ELECTION_YAML}\nreplication:\n  reseed_self_restart: true");
    let mut cluster = MultiProcCluster::start_sharded(3, GROUPS, Some(&topology));
    let leaders = cluster.wait_shard_leaders_agreed(convergence_budget());
    println!("[multi] initial group leaders: {leaders:?}");

    // Baseline across every group. Entity ids are hash-routed, so a spread of ids
    // lands writes in all three groups.
    for entity in 1..=30u64 {
        write_heavy_item_retrying(&cluster, LEADER, entity, "", false);
    }
    cluster.wait_converged_all(convergence_budget());
    println!("[multi] baseline converged on all three nodes");

    // Node 2 leaves. Its group-2 leadership moves, so a survivor now leads two
    // groups — exactly the live tidaldb-1 arrangement.
    //
    // `wait_shard_leaders_agreed` only requires the LIVE nodes to AGREE, and they
    // agree on the dead node until the election timeout elapses. Writes are
    // hash-routed, so until group 2's leadership actually moves, every entity
    // landing in that group has no live leader and `/items` is not a 201. Wait for
    // the handoff itself, which is also the arrangement being reproduced.
    cluster.stop_graceful(AP_SOUTH);
    let stopped = cluster.region_name(AP_SOUTH).to_string();
    let handoff_deadline = Instant::now() + convergence_budget();
    let after = loop {
        let map = cluster.wait_shard_leaders_agreed(convergence_budget());
        if map.values().all(|l| *l != stopped) {
            break map;
        }
        assert!(
            Instant::now() < handoff_deadline,
            "group leadership never vacated the stopped node {stopped}: {map:?}"
        );
        std::thread::sleep(Duration::from_millis(250));
    };
    assert!(
        after
            .values()
            .any(|l| after.values().filter(|x| *x == l).count() > 1),
        "expected a survivor to lead TWO groups after the handoff (the live arrangement), \
         got {after:?}"
    );
    println!("[multi] node 2 down; group leaders now {after:?} (a survivor leads two groups)");

    // Enough heavy writes to rotate past WAL_RETENTION_SEGMENTS on the groups the
    // survivors lead, so node 2's resume points are genuinely compacted away.
    let blob = blob_value();
    for entity in 31..=(30 + OFFLINE_ITEMS) {
        write_heavy_item_retrying(&cluster, LEADER, entity, &blob, true);
    }
    println!("[multi] wrote {OFFLINE_ITEMS} heavy items while node 2 was down");

    // Graceful restarts of the live nodes run the clean-shutdown compaction.
    cluster.restart_graceful(LEADER, &[]);
    cluster.restart_graceful(EU_WEST, &[]);
    let post = cluster.wait_shard_leaders_agreed(convergence_budget());
    println!("[multi] survivors gracefully restarted (WAL compacted); leaders {post:?}");

    // Node 2 returns needing a reseed on more than one group.
    cluster.restart(AP_SOUTH, &[]);

    // FIXPOINT, with the test standing in for the ORCHESTRATOR.
    //
    // `reseed_self_restart` drains and exits(0) expecting something to boot the
    // process again — in production that is the kubelet's restart policy. The
    // harness has no supervisor: an exited node simply stays down and
    // `is_alive` still reports true (it only checks that the handle is retained).
    // So this loop IS the supervisor: sustained HTTP unreachability is the exit
    // signal, and `restart` (which sigkills idempotently first) is the reboot.
    // Counting those reboots is the gate — one per group needing a reseed is the
    // designed cost; unbounded is the live loop.
    let deadline = Instant::now() + convergence_budget() + Duration::from_secs(240);
    let mut restarts: u32 = 0;
    let mut unreachable_polls = 0u32;
    loop {
        match cluster.local_status(AP_SOUTH) {
            Some(s) => {
                unreachable_polls = 0;
                // EVERY hosted group must be settled, not just the flat fields.
                // The flat `reseed_required` / `reseeding` describe only the group
                // `replica_for(None)` resolves (the lowest hosted id), so on a
                // 3-group node they can read "clean" while another group is still
                // marked. That under-reporting is exactly what made the production
                // incident look converged, and it fooled the first version of this
                // predicate too — the node announced "settled after 0 restarts" and
                // was missing an item. The per-group rows now carry the state.
                let rows = s["shards"].as_array().cloned().unwrap_or_default();
                let all_groups_clean = !rows.is_empty()
                    && rows.iter().all(|r| {
                        r["reseed_required"].as_bool() == Some(false)
                            && r["reseeding"].as_bool() == Some(false)
                    });
                if all_groups_clean
                    && s["reseed_required"].as_bool() == Some(false)
                    && s["reseeding"].as_bool() == Some(false)
                    && s["quarantined"].as_bool() == Some(false)
                {
                    println!(
                        "[multi] node 2 settled after {restarts} orchestrator restart(s); \
                         per-group rows: {rows:?}"
                    );
                    break;
                }
            }
            None => {
                unreachable_polls += 1;
                // ~2s unreachable = it drained and exited (a self-restart), not a
                // momentary blip. Reboot it exactly as the kubelet would.
                if unreachable_polls >= 4 {
                    unreachable_polls = 0;
                    restarts += 1;
                    assert!(
                        restarts <= RESTART_CEILING,
                        "node 2 has been rebooted {restarts} times (ceiling {RESTART_CEILING}) \
                         without settling: a multi-group node is healing at most one group per \
                         boot and never reaching a fixpoint. This is the 2026-08-20 production \
                         loop, where the exiting group alternated 2 -> 1 -> 2 across 8 restarts \
                         in 22 minutes while re-streaming a full snapshot each cycle."
                    );
                    println!(
                        "[multi] node 2 exited (self-restart); orchestrator reboot #{restarts}"
                    );
                    cluster.restart(AP_SOUTH, &[]);
                }
            }
        }
        assert!(
            Instant::now() < deadline,
            "node 2 never settled; reboots={restarts}, last status: {:?}",
            cluster.local_status(AP_SOUTH)
        );
        std::thread::sleep(Duration::from_millis(500));
    }

    // CONTENT across every group: a settled marker means nothing if the log has a
    // hole. Probe ids spanning the hash space so all three groups are exercised.
    //
    // STILL SUPERVISED. A node can settle and then re-latch and exit again — the
    // first run of this fixture did exactly that, and an unsupervised probe just
    // panicked on a connection error, hiding the interesting behaviour. Keep
    // rebooting against the SAME ceiling here, so a node that only appears to
    // converge is still caught, and use `local_status` (an Option) rather than a
    // bare GET so unreachability is data instead of a panic.
    let probe_deadline = Instant::now() + convergence_budget() + Duration::from_secs(120);
    for entity in [31u64, 100, 500, 1500, 30 + OFFLINE_ITEMS] {
        loop {
            if cluster.local_status(AP_SOUTH).is_some() {
                unreachable_polls = 0;
                if item_searchable(&cluster, AP_SOUTH, entity) {
                    break;
                }
            } else {
                unreachable_polls += 1;
                if unreachable_polls >= 4 {
                    unreachable_polls = 0;
                    restarts += 1;
                    assert!(
                        restarts <= RESTART_CEILING,
                        "node 2 settled and then RE-LATCHED and exited again, reboot {restarts} \
                         of ceiling {RESTART_CEILING}: convergence was not a fixpoint. This is \
                         the production shape — apparent convergence followed by another \
                         self-restart cycle."
                    );
                    println!(
                        "[multi] node 2 exited AFTER settling; orchestrator reboot #{restarts}"
                    );
                    cluster.restart(AP_SOUTH, &[]);
                }
            }
            assert!(
                Instant::now() < probe_deadline,
                "reseeded multi-group node is missing item {entity} (reboots={restarts}): the \
                 groups settled but the data did not arrive, which is the silent-hole shape this \
                 suite must never pass. status: {:?}",
                cluster.local_status(AP_SOUTH)
            );
            std::thread::sleep(Duration::from_millis(250));
        }
    }
    println!(
        "[multi] every probed item is readable on the reseeded multi-group node \
         ({restarts} reboot(s) total)"
    );
}

/// POST with the `x-tidal-ack` header through a dedicated client. `Some(seq)`
/// only for a 2xx carrying `x-tidal-seq` (the ledger's "acknowledged").
fn post_acked(
    client: &reqwest::blocking::Client,
    base: &str,
    path: &str,
    ack: &str,
    body: &serde_json::Value,
) -> Option<u64> {
    let resp = client
        .post(format!("{base}{path}"))
        .header("x-tidal-ack", ack)
        .json(body)
        .send()
        .ok()?;
    if !resp.status().is_success() {
        return None;
    }
    resp.headers()
        .get("x-tidal-seq")?
        .to_str()
        .ok()?
        .parse()
        .ok()
}

/// Poll the LIVE candidate nodes for an elected leader at a term above
/// `after_term`. Returns `(node_idx, term)`.
fn await_elected_leader(
    cluster: &MultiProcCluster,
    candidates: &[usize],
    after_term: u64,
    budget: Duration,
) -> (usize, u64) {
    let deadline = Instant::now() + budget;
    loop {
        for &idx in candidates {
            if let Some(status) = cluster.local_status(idx) {
                let term = status["term"].as_u64().unwrap_or(0);
                if status["role"].as_str() == Some("leader") && term > after_term {
                    return (idx, term);
                }
            }
        }
        assert!(
            Instant::now() < deadline,
            "no leader elected among {candidates:?} above term {after_term} within {budget:?}"
        );
        std::thread::sleep(Duration::from_millis(50));
    }
}

/// Gate 3: a quarantined divergent node reseeds via the boot-time snapshot
/// install — NO `wipe_data_dir`. Cribbed from `cluster_election.rs`'s
/// quarantine drill (`FAST_ELECTION_YAML`; a divergent suffix exists; survivors
/// elect; restart the old leader → it quarantines), then closes the loop the
/// m11p4 drill left to a wipe: the quarantine latch now writes the reseed
/// marker, and a second restart marker-boots a snapshot install from the
/// CURRENT ELECTED leader.
///
/// The divergent suffix is made DETERMINISTIC (not timing-probabilistic) by
/// severing the leader's OUTBOUND gRPC ship to both followers BEFORE the
/// suffix writes: those `ack=leader` writes are durably journaled on the
/// leader but never reach a follower, so the survivors — starved of the
/// leader's heartbeats — election-timeout and elect a new leader whose
/// election-time position EXCLUDES the suffix. When the old leader restarts
/// and joins the new term, the §5 divergence check sees its WAL tail exceed
/// the new leadership's history → quarantine.
#[test]
fn mp_quarantined_node_reseeds_without_wipe() {
    use support::partition::{ProxyController, proxied_rewrite};

    /// Sever the leader's outbound gRPC to a follower (its own view of the
    /// peer): cuts ship + heartbeat so the suffix never replicates and the
    /// follower's election timer is no longer reset.
    fn sever_ship(proxies: &ProxyController, leader: &str, follower: &str) {
        proxies.edge(leader, follower).grpc.sever();
    }

    let regions = [
        cluster_region_name(LEADER),
        cluster_region_name(EU_WEST),
        cluster_region_name(AP_SOUTH),
    ];
    let region_refs: Vec<&str> = regions.iter().map(String::as_str).collect();
    let (rewrite, proxies) = proxied_rewrite(&region_refs);
    // Give the (to-be-quarantined) leader region the LOW handshake window so its
    // eventual marker boot installs inside budget.
    let opts = ClusterOptions::new(3)
        .with_topology_extra(FAST_ELECTION_YAML)
        .with_rewrite(rewrite)
        .with_env(LEADER, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS);
    let mut cluster = MultiProcCluster::start_with(opts);
    let client = reqwest::blocking::Client::builder()
        .timeout(Duration::from_secs(3))
        .build()
        .unwrap();

    // A committed prefix every node shares (quorum-acked while links are intact).
    for entity in 1..=5u64 {
        post_acked(
            &client,
            &cluster.node(LEADER),
            "/items",
            "quorum",
            &serde_json::json!({ "entity_id": entity, "metadata": { "title": item_token(entity) } }),
        )
        .expect("committed-prefix quorum write");
    }
    cluster.wait_converged_all(convergence_budget());

    // ── Sever the leader's OUTBOUND ship to both followers, then write a burst
    // of `ack=leader` items: durably journaled on the leader, never shipped —
    // the deterministic DIVERGENT SUFFIX.
    let leader_region = cluster_region_name(LEADER);
    sever_ship(&proxies, &leader_region, &cluster_region_name(EU_WEST));
    sever_ship(&proxies, &leader_region, &cluster_region_name(AP_SOUTH));
    let mut suffix_writes = 0u32;
    let burst_deadline = Instant::now() + Duration::from_millis(700);
    let mut entity = 100u64;
    while Instant::now() < burst_deadline {
        if post_acked(
            &client,
            &cluster.node(LEADER),
            "/items",
            "leader",
            &serde_json::json!({ "entity_id": entity, "metadata": { "title": item_token(entity) } }),
        )
        .is_some()
        {
            suffix_writes += 1;
        }
        entity += 1;
    }
    assert!(
        suffix_writes > 0,
        "the severed leader must have journaled an unshipped divergent suffix"
    );
    println!("[quarantine] severed leader ship; journaled {suffix_writes} unshipped suffix writes");

    // SIGKILL the leader (the suffix stays on its WAL, unshipped) and stop the
    // writer.
    cluster.kill_hard(LEADER);

    // ── The survivors elect on their own (zero operator verbs). They never saw
    // the suffix, so the elected leader's position excludes it.
    let survivors = [EU_WEST, AP_SOUTH];
    let (new_leader, new_term) =
        await_elected_leader(&cluster, &survivors, 0, Duration::from_secs(15));
    // Writes resume on the new leadership so the elected term is real and
    // advancing (the reseed needs a live leader to fetch from).
    let resume_deadline = Instant::now() + Duration::from_secs(10);
    loop {
        if post_acked(
            &client,
            &cluster.node(new_leader),
            "/signals",
            "quorum",
            &serde_json::json!({ "entity_id": 900_001, "signal": "view", "weight": 1.0 }),
        )
        .is_some()
        {
            break;
        }
        assert!(
            Instant::now() < resume_deadline,
            "quorum writes did not resume on the elected leader"
        );
        std::thread::sleep(Duration::from_millis(50));
    }
    println!("[quarantine] survivors elected node {new_leader} at term {new_term}; writes resumed");

    // ── Restart the old leader with its OUTBOUND catch-up to the followers
    // STILL severed (the `edge(leader, follower)` links from the suffix step).
    //
    // Why this is the load-bearing determinism, not an artifice: the divergence
    // quarantine is detected by `join_term_check`, which runs ONLY on a
    // heartbeat carrying the new leader's election-time position and compares it
    // against this node's `election_log_position()`. That position reads the
    // node's OWN term-0 suffix frontier ONLY while its WAL tail term is still 0.
    // If the node's INDEPENDENT catch-up pull applies the new leader's term-N
    // marker FIRST, the tail term flips to N and the position reads the new
    // stream's (lower) applied frontier instead — masking the divergent suffix,
    // so the node silently converges and never quarantines (a real race, also
    // why `cluster_election.rs` treats quarantine as conditional). Heartbeats are
    // the leader DIALING us (`edge(follower, leader)`, intact); catch-up is US
    // dialing the leader (`edge(leader, follower)`, severed). Keeping the
    // OUTBOUND severed lets the heartbeat-driven check run against the intact
    // term-0 suffix → DETERMINISTIC quarantine. The links heal before the
    // marker boot, which needs to dial the leader for discovery + snapshot.
    cluster.restart(LEADER, &[]);
    await_status_bool(
        &cluster,
        LEADER,
        "quarantined",
        true,
        convergence_budget() + Duration::from_secs(10),
        "the restarted old leader quarantines on its divergent suffix",
    );
    // The quarantine flag (election_driver `quarantined.store`) and the reseed
    // marker (`latch_reseed_marker`) are latched SEQUENTIALLY, not atomically, so
    // there is a brief window where status reports quarantined=true before the
    // marker surfaces. Poll for the marker (matching the quarantined check above)
    // rather than a single-shot read, which races under concurrent-test load.
    await_status_bool(
        &cluster,
        LEADER,
        "reseed_required",
        true,
        convergence_budget() + Duration::from_secs(10),
        "the quarantine latch must also write the reseed marker (m11p5 §2.4)",
    );
    println!("[quarantine] old leader quarantined AND latched reseed_required (no wipe)");

    // ── Heal the leader's outbound so the marker boot can dial the elected
    // leader (HTTP status discovery + the gRPC snapshot fetch).
    proxies
        .edge(&leader_region, &cluster_region_name(EU_WEST))
        .grpc
        .heal_link();
    proxies
        .edge(&leader_region, &cluster_region_name(AP_SOUTH))
        .grpc
        .heal_link();

    // ── Restart it ONCE MORE → marker boot → snapshot install from the CURRENT
    // ELECTED leader (the discovery loop polls peers for is_leader, adopts the
    // elected term, fetches, swaps). NO wipe_data_dir anywhere.
    cluster.restart(LEADER, &[]);
    await_self_converged(
        &cluster,
        LEADER,
        convergence_budget() + Duration::from_secs(20),
        "the marker boot installs the snapshot from the elected leader and converges",
    );
    println!("[quarantine] old leader marker-booted, reseeded from the elected leader, converged");

    // ── Clean rejoin: serves reads, no quarantine, no marker. The process-local
    // divergence gauge is cleared by the fresh process + the install removing
    // the divergent suffix (so the next term-join does not re-quarantine); the
    // `quarantined` status field mirrors that gauge — both fresh-false after the
    // reseed boot. No `/metrics` port is wired in the harness, so the status
    // field is the in-budget proxy for the gauge (it cannot read `false` while
    // the gauge reads `true` — they are set from the same code path).
    assert!(
        !status_bool(&cluster, LEADER, "quarantined"),
        "the reseeded node must clear its quarantine: {:?}",
        cluster.local_status(LEADER)
    );
    assert!(
        !status_bool(&cluster, LEADER, "reseed_required"),
        "the reseeded node must clear its reseed marker: {:?}",
        cluster.local_status(LEADER)
    );

    // It serves reads (an item the elected leader wrote during/after the kill is
    // visible on the reseeded node — the snapshot+stream delivered it).
    // Convergence-class wait: the snapshot install + text-index rebuild +
    // ~2 s auto-commit compete with suite-parallel load, so a fixed short
    // budget flakes on busy hosts; use the harness convergence budget
    // (30 s default, TIDAL_TEST_CONVERGENCE_BUDGET_SECS to widen).
    let probe_deadline = Instant::now() + convergence_budget();
    loop {
        // Item 1 was written by the original leader's load; it is part of the
        // committed prefix the elected leader carried, so the reseeded node must
        // serve it.
        if item_searchable(&cluster, LEADER, 1) {
            break;
        }
        assert!(
            Instant::now() < probe_deadline,
            "the reseeded node does not serve a committed item — reads are broken"
        );
        std::thread::sleep(Duration::from_millis(200));
    }
    println!("[quarantine] reseeded node serves reads; quarantined=false, reseed_required=false");
}

/// Production-readiness roadmap, task 00 — the Ring-0 foundation.
///
/// A healthy, caught-up follower must NOT reseed across a GRACEFUL ROLLING
/// RESTART that moves leadership to another region. This reproduces the
/// production blocker: every rolling deploy / upgrade / node reboot currently
/// churns the cluster through a `from_seqno=1` snapshot reseed-on-rejoin even
/// though no data was lost and the node was caught up at SIGTERM.
///
/// The repro deliberately drives BOTH conditions the live symptom needs:
///
/// * **a corpus past compaction** — the same heavy [`OFFLINE_ITEMS`] batch the
///   sibling reseed proof uses, so the graceful-shutdown compaction deletes the
///   early WAL and a `from_seqno=1` pull is *genuinely* unservable (the exact
///   `WAL compacted below seqno 1` refusal), not merely suboptimal; and
/// * **a leadership move across the restart** — elected era + restarting the
///   current leader first, so the within-term `own.tail_term == term`
///   clean-rejoin shortcut (`election_driver.rs`) cannot hide a join
///   misclassification.
///
/// It then asserts no node latches `reseed_required` and every acked item still
/// reads back. The reseed marker is durable and `reseed_self_restart` defaults
/// off, so a wrongful latch STAYS latched — `await_status_bool(.., false, ..)`
/// times out on the bug rather than racing a self-heal.
///
/// FAILS today — that failure IS the deliverable of task 00 (it pins which of
/// Bugs A/B/C fire and guards tasks 01–03). Greens once task 01 (boot self-heal
/// resume floor) + task 02 (`decide_join` numbering) land.
#[test]
fn mp_graceful_rolling_restart_preserves_applied_no_reseed() {
    // Elected era (leadership can move on restart) + fast timers (the move and
    // the rejoin both land inside the test budget). Every node gets the LOW
    // handshake window so that IF a reseed wrongly latches, the test still
    // bounds; the assertion is that it must NOT.
    let opts = ClusterOptions::new(3)
        .with_topology_extra(PROD_ELECTION_YAML)
        .with_env(LEADER, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS)
        .with_env(EU_WEST, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS)
        .with_env(AP_SOUTH, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS);
    let mut cluster = MultiProcCluster::start_with(opts);
    let _ = cluster.wait_shard_leaders_agreed(convergence_budget());

    let seed_leader = current_leader_idx(&cluster).expect("an elected leader before seeding");

    // Baseline prefix shared by all three, then the heavy batch that pushes the
    // WAL well past the 16-segment retention window so the graceful-shutdown
    // compaction genuinely deletes the early segments.
    // The mechanism is term-marker / uncommitted-tail divergence on step-down, NOT
    // compaction (the repro disambiguated this), so a modest converged corpus is
    // enough — it just needs leadership to be able to move across the restarts.
    const ROLLING_SEED: u64 = 400;
    for entity in 1..=ROLLING_SEED {
        write_heavy_item_only(&cluster, seed_leader, entity, "");
    }
    cluster.wait_converged_all(convergence_budget() + Duration::from_secs(10));
    println!("[rolling] seeded {ROLLING_SEED} items; all three converged");

    // Match production: a long-running cluster is always elected-era. Force the
    // term-0 → elected transition once up front so the measured restarts below
    // exercise only the elected-era step-down path.
    ensure_elected_era(&mut cluster);
    println!("[rolling] cluster forced into the elected era");

    // Converged steady state: nobody is reseed-pending.
    for idx in 0..3 {
        await_status_bool(
            &cluster,
            idx,
            "reseed_required",
            false,
            convergence_budget(),
            "no reseed latched in the converged steady state",
        );
    }

    // ── GRACEFUL ROLLING RESTART, one node at a time, in POD-ORDINAL order
    // (0,1,2) — exactly what a k8s `RollingUpdate` does. Whichever ordinal is the
    // current leader moves leadership exactly once when its turn comes; a follower
    // restart leaves leadership put (production timers keep the lease). Every node
    // fully rejoins and the cluster re-agrees a leader before the next is taken
    // down. The deposed leader rejoining clean (no reseed) is the whole point.
    let order = [0usize, 1, 2];
    let mut leadership_moved = false;

    for &node in &order {
        let pre = cluster
            .region_name(current_leader_idx(&cluster).expect("a leader before restart"))
            .to_string();
        cluster.restart_graceful(node, &[]);
        // The cluster must re-agree a leader for every shard group. Reaching
        // agreement requires the just-restarted node to have processed the
        // current leader's heartbeat — i.e. its `join_term_check` has already
        // run — so any wrongful reseed latch is durable by the time we poll it.
        let _ = cluster.wait_shard_leaders_agreed(convergence_budget() + Duration::from_secs(20));
        let post = cluster
            .region_name(current_leader_idx(&cluster).expect("a leader after restart"))
            .to_string();
        if post != pre {
            leadership_moved = true;
        }

        // THE INVARIANT: no node latches reseed across a graceful restart. A
        // wrongful latch is durable (no self-restart), so this times out on the
        // bug — it does not race a self-heal.
        for idx in 0..3 {
            await_status_bool(
                &cluster,
                idx,
                "reseed_required",
                false,
                convergence_budget() + Duration::from_secs(10),
                "a graceful rolling restart must not latch reseed",
            );
            // Applied frontier preserved (a snapshot reseed momentarily drops the
            // node's applied position; a clean stream-catch-up never does).
            let applied = cluster
                .local_status(idx)
                .and_then(|s| s["applied_events"].as_u64())
                .unwrap_or(0);
            assert!(
                applied > 0,
                "node {idx} applied frontier must be preserved across the restart \
                 (a reseed-from-1 resets it): {:?}",
                cluster.local_status(idx)
            );
        }
        // Pace like a readiness-gated k8s rollout: do not take down the next node
        // until the cluster has fully re-converged (idle ⇒ strict lag 0).
        await_cluster_caught_up(
            &cluster,
            0,
            convergence_budget() + Duration::from_secs(20),
            "cluster re-converges before the next rolling restart",
        );
        println!("[rolling] restarted node {node}; leader {pre} -> {post}; no reseed on any node");
    }
    assert!(
        leadership_moved,
        "the rolling restart MUST move leadership at least once, else the \
         within-term clean-rejoin shortcut hides the join misclassification"
    );

    // ── Zero acked-write loss: known seeded ids still read back on every node.
    let probe_deadline = Instant::now() + convergence_budget();
    for entity in [1u64, 4, 200, ROLLING_SEED] {
        for idx in 0..3 {
            loop {
                if item_searchable(&cluster, idx, entity) {
                    break;
                }
                assert!(
                    Instant::now() < probe_deadline,
                    "node {idx} lost item {entity} across the rolling restart \
                     (zero acked-write loss is violated)"
                );
                std::thread::sleep(Duration::from_millis(200));
            }
        }
    }
    println!("[rolling] every probed item readable on all three nodes — zero loss, zero reseed");
}

/// Production-readiness roadmap, task 04 — the graceful rolling restart UNDER
/// SUSTAINED WRITE LOAD must not reseed any node. This is the faithful production
/// scenario the nightly soak exercises: writes are in flight when a pod is
/// recycled, so the leader carries an UNCOMMITTED tail at SIGTERM. Without the
/// graceful leadership hand-off (drain-to-quorum before step-down) that tail
/// diverges from the next term and the deposed leader reseeds on rejoin — the
/// per-rollout churn. With the hand-off, the tail commits first and every node
/// rejoins clean. Production election timers + k8s ordinal restart order.
#[test]
fn mp_graceful_rolling_restart_under_load_no_reseed() {
    let opts = ClusterOptions::new(3)
        .with_topology_extra(PROD_ELECTION_YAML)
        .with_env(LEADER, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS)
        .with_env(EU_WEST, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS)
        .with_env(AP_SOUTH, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS);
    let mut cluster = MultiProcCluster::start_with(opts);
    let _ = cluster.wait_shard_leaders_agreed(convergence_budget());

    // Modest converged baseline — this test exercises the DECISION under load, not
    // scale (keep it fast; the corpus need not force compaction).
    let seed_leader = current_leader_idx(&cluster).expect("a leader");
    for entity in 1..=200u64 {
        write_heavy_item_only(&cluster, seed_leader, entity, "");
    }
    cluster.wait_converged_all(convergence_budget());
    ensure_elected_era(&mut cluster); // production is always elected-era
    let known = 42u64; // a committed baseline id we prove still serves at the end
    println!("[under-load] baseline of 200 items converged (elected era); starting the writer");

    // ── Background writer: steady item writes round-robined across the nodes (the
    // gateway forwards to the current leader), tolerating the transient failures a
    // restart causes. The point is sustained in-flight load so the leader always
    // has an uncommitted tail when its pod is recycled. Acked (201) ids are
    // recorded so we can prove the cluster keeps serving committed content.
    let urls: Vec<String> = (0..3).map(|i| cluster.node(i)).collect();
    let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    let acked = std::sync::Arc::new(std::sync::Mutex::new(Vec::<u64>::new()));
    let writer = {
        let stop = std::sync::Arc::clone(&stop);
        let acked = std::sync::Arc::clone(&acked);
        std::thread::spawn(move || {
            let client = reqwest::blocking::Client::builder()
                .timeout(Duration::from_secs(2))
                .build()
                .expect("writer client");
            let mut id = 2_000_000u64;
            let mut rr = 0usize;
            while !stop.load(std::sync::atomic::Ordering::Relaxed) {
                let url = format!("{}/items", urls[rr % urls.len()]);
                rr += 1;
                let body = serde_json::json!({
                    "entity_id": id,
                    "metadata": { "title": item_token(id) },
                });
                // ack=quorum: a 2xx means the write COMMITTED to a quorum, so it can
                // never become a divergent suffix on a leadership change (unlike the
                // leader-only ack, whose un-replicated tail is correctly truncated).
                // This is the durability contract a cluster that rolling-restarts
                // under load must use; the test proves a committed write is never lost
                // and never triggers a reseed.
                if let Ok(resp) = client
                    .post(&url)
                    .header("x-tidal-ack", "quorum")
                    .json(&body)
                    .send()
                    && resp.status().is_success()
                {
                    acked.lock().unwrap().push(id);
                }
                id += 1;
                std::thread::sleep(Duration::from_millis(20)); // ~50 writes/s
            }
        })
    };

    // ── Ordinal rolling restart UNDER LOAD; assert no reseed after each node.
    let mut leadership_moved = false;
    for node in [0usize, 1, 2] {
        let pre = cluster
            .region_name(current_leader_idx(&cluster).expect("a leader"))
            .to_string();
        cluster.restart_graceful(node, &[]);
        let _ = cluster.wait_shard_leaders_agreed(convergence_budget() + Duration::from_secs(20));
        let post = cluster
            .region_name(current_leader_idx(&cluster).expect("a leader"))
            .to_string();
        if post != pre {
            leadership_moved = true;
        }
        for idx in 0..3 {
            await_status_bool(
                &cluster,
                idx,
                "reseed_required",
                false,
                convergence_budget() + Duration::from_secs(10),
                "a graceful rolling restart UNDER LOAD must not latch reseed",
            );
        }
        // Pace like a readiness-gated rollout: let the cluster re-converge (modulo
        // the small in-flight write tail) before taking down the next node, so the
        // next leader's step-down drain has caught-up followers to commit against.
        await_cluster_caught_up(
            &cluster,
            50,
            convergence_budget() + Duration::from_secs(20),
            "cluster re-converges (under load) before the next rolling restart",
        );
        println!("[under-load] restarted node {node}; leader {pre} -> {post}; no reseed");
    }
    assert!(
        leadership_moved,
        "leadership must move at least once across the rolling restart"
    );

    // Stop the writer; it must have landed a meaningful amount of acked load.
    stop.store(true, std::sync::atomic::Ordering::Relaxed);
    writer.join().expect("writer thread joins");
    let ids = acked.lock().unwrap().clone();
    // A modest floor: writes are deliberately in flight across the restarts, but
    // each failover briefly has no write-accepting leader (and a shutting-down
    // leader rejects writes — the quiesce that lets the drain converge), and a
    // forwarded write can burn its 2s timeout chasing a moving leader, so the
    // ACKED count is naturally low. The point is only that real load WAS flowing.
    assert!(
        !ids.is_empty(),
        "the writer must have landed some acked load under the restarts (got {})",
        ids.len()
    );
    println!(
        "[under-load] writer landed {} acked writes under the restarts",
        ids.len()
    );

    // ── The committed corpus still SERVES on every node (no reseed wiped it; the
    // cluster stayed correct through the load + restarts). A committed baseline id
    // must read back everywhere.
    let probe_deadline = Instant::now() + convergence_budget();
    for idx in 0..3 {
        loop {
            if item_searchable(&cluster, idx, known) {
                break;
            }
            assert!(
                Instant::now() < probe_deadline,
                "node {idx} no longer serves committed item {known} after the under-load \
                 rolling restart"
            );
            std::thread::sleep(Duration::from_millis(200));
        }
    }
    println!("[under-load] committed corpus still serves on all nodes — clean under load");
}

/// One node's election-divergence-relevant status, sampled for the Ring-0
/// consistency oracle. `election_frontier`/`election_tail_term` are the LIVE
/// `election_log_position()` the vote restriction + `decide_join` read;
/// `applied_events` is this node's applied frontier in the CURRENT leader's
/// stream numbering. The consistency invariant: a caught-up follower
/// (`lag_events == 0`, not reseeding) reports `election_frontier ==
/// applied_events` — i.e. the position it would VOTE/JOIN with is the same
/// quantity, in the same numbering, the leader holds. The cross-numbering tear
/// (an ex-leader still reading its OWN `flushed_seq` because its `wal_term_mark`
/// has not yet folded the new leader's marker) shows up as `election_frontier !=
/// applied_events` on a node that is otherwise caught up, and/or as a spurious
/// `quarantined`.
// Fields `idx`/`term`/`election_tail_term` are diagnostic-only: they surface in
// the `{violations:#?}` dump when the oracle fails (dead-code analysis ignores
// `Debug`), so silence the lint rather than drop the evidence.
#[allow(dead_code)]
#[derive(Debug, Clone)]
struct ElectionSample {
    idx: usize,
    is_leader: bool,
    quarantined: bool,
    reseeding: bool,
    term: u64,
    election_tail_term: u64,
    election_frontier: u64,
    applied_events: u64,
    lag_events: u64,
}

fn election_sample(cluster: &MultiProcCluster, idx: usize) -> Option<ElectionSample> {
    let s = cluster.local_status(idx)?;
    Some(ElectionSample {
        idx,
        is_leader: s["is_leader"].as_bool().unwrap_or(false),
        quarantined: s["quarantined"].as_bool().unwrap_or(false),
        reseeding: s["reseeding"].as_bool().unwrap_or(false),
        term: s["term"].as_u64().unwrap_or(0),
        election_tail_term: s["election_tail_term"].as_u64().unwrap_or(u64::MAX),
        election_frontier: s["election_frontier"].as_u64().unwrap_or(u64::MAX),
        applied_events: s["applied_events"].as_u64().unwrap_or(0),
        lag_events: s["lag_events"].as_u64().unwrap_or(u64::MAX),
    })
}

/// Ring 0 (election-divergence-fix roadmap, task 00) — the CONSISTENCY ORACLE.
///
/// Pins the invariant the whole roadmap protects: for the same committed state,
/// every node's reported election position is the SAME comparable quantity in
/// ONE numbering, and a CAUGHT-UP node is never misclassified as divergent.
///
/// It forces an elected-era failover (graceful restart of the current leader),
/// then SAMPLES every node continuously through the rejoin until the cluster
/// re-converges, asserting two things on every sample:
///   1. No caught-up node (`lag_events == 0`, not reseeding) is `quarantined`
///      — a caught-up node has nothing the new leadership does not subsume.
///   2. A caught-up non-leader's `election_frontier == applied_events` — the
///      position it would vote/join with IS its applied frontier in the leader's
///      stream (the cross-numbering tear violates this).
///
/// FAILS today under the cross-numbering tear (the deposed leader reads its own
/// `flushed_seq` for `election_frontier` while `applied_events` is the leader
/// stream — they differ, and/or it false-quarantines). Greens once task 01
/// (stream-numbered marker) + task 02 (`election_log_position_for`) + task 03
/// (committed-subsumption) make the position consistent. The negative controls
/// (`mp_quarantined_*`, `mp_follower_*`) keep reseeding — genuine divergence is
/// untouched.
///
/// Greens with the m12 election-divergence-fix (the durable `leader_acked`
/// frontier): a caught-up node is never false-quarantined across an elected-era
/// failover, and the position is consistent at rest. The negative controls
/// (`mp_quarantined_*`, `mp_follower_*`) keep reseeding.
#[test]
fn mp_election_position_consistent_across_roles_after_failover() {
    let opts = ClusterOptions::new(3)
        .with_topology_extra(PROD_ELECTION_YAML)
        .with_env(LEADER, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS)
        .with_env(EU_WEST, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS)
        .with_env(AP_SOUTH, "TIDAL_RESEED_HANDSHAKE_MS", RESEED_HANDSHAKE_MS);
    let mut cluster = MultiProcCluster::start_with(opts);
    let _ = cluster.wait_shard_leaders_agreed(convergence_budget());

    let seed_leader = current_leader_idx(&cluster).expect("an elected leader before seeding");
    for entity in 1..=200u64 {
        write_heavy_item_only(&cluster, seed_leader, entity, "");
    }
    cluster.wait_converged_all(convergence_budget() + Duration::from_secs(10));
    ensure_elected_era(&mut cluster); // production is always elected-era
    println!("[oracle] 200-item baseline converged (elected era)");

    // Converged steady state: assert the consistency invariant holds at rest on
    // every node BEFORE any failover (the baseline the tear later violates).
    await_cluster_caught_up(
        &cluster,
        0,
        convergence_budget(),
        "cluster converges before the measured failover",
    );
    for idx in 0..3 {
        if let Some(s) = election_sample(&cluster, idx) {
            assert!(
                !s.quarantined,
                "[oracle] node {idx} quarantined in the converged steady state: {s:?}"
            );
        }
    }

    // ── Force a failover: graceful-restart the current leader. Survivors elect a
    // new term; the deposed leader reboots as a follower and rejoins. THIS rejoin
    // is where the cross-numbering tear fires (the deposed leader's `wal_term_mark`
    // still names itself until it folds the new leader's marker).
    let deposed = current_leader_idx(&cluster).expect("a leader to depose");
    let pre = cluster.region_name(deposed).to_string();
    cluster.restart_graceful(deposed, &[]);

    // SAMPLE THE REJOIN WINDOW. Poll every node ~10×/s until the cluster
    // re-converges (all lag 0) or the budget elapses, asserting the invariant on
    // every sample. A violation captures the exact cross-numbering evidence.
    let deadline = Instant::now() + convergence_budget() + Duration::from_secs(30);
    // THE ROBUST INVARIANT, sampled continuously through the rejoin: a genuinely
    // caught-up node is NEVER quarantined. "Genuinely caught up" requires a SEEDED
    // lag gauge (`applied_events > 0`), not a bare `lag == 0` — right after a
    // failover an UNINITIALIZED gauge reads `leader_seqno(0) - applied(0) = 0`, a
    // spurious "caught up, applied nothing". `quarantined` is a durable latch
    // immune to that window. This false-quarantine is exactly what the fix removes.
    let mut quarantine_violations: Vec<ElectionSample> = Vec::new();
    let mut converged = false;
    while Instant::now() < deadline {
        let samples: Vec<ElectionSample> = (0..3)
            .filter_map(|i| election_sample(&cluster, i))
            .collect();
        for s in &samples {
            let caught_up = s.lag_events == 0 && s.applied_events > 0 && !s.reseeding;
            if caught_up && s.quarantined {
                quarantine_violations.push(s.clone());
            }
        }
        // GENUINE re-convergence: every node has a SEEDED gauge at lag 0 and a
        // leader exists — past the post-failover uninitialized-gauge window.
        if samples.len() == 3
            && samples
                .iter()
                .all(|s| s.lag_events == 0 && s.applied_events > 0)
            && samples.iter().any(|s| s.is_leader)
        {
            converged = true;
            break;
        }
        std::thread::sleep(Duration::from_millis(100));
    }
    let post = current_leader_idx(&cluster)
        .map(|i| cluster.region_name(i).to_string())
        .unwrap_or_else(|| "<none>".into());
    println!("[oracle] failover {pre} -> {post}; converged={converged}");

    assert!(
        quarantine_violations.is_empty(),
        "[oracle] FALSE QUARANTINE across the failover ({} sample(s)) — a genuinely \
         caught-up node was quarantined (the cross-numbering false-quarantine the \
         fix removes). Evidence: {quarantine_violations:#?}",
        quarantine_violations.len()
    );
    assert!(
        converged,
        "[oracle] cluster did not genuinely re-converge after the failover (leader \
         {pre} -> {post})"
    );

    // AT GENUINE REST, the CONSISTENCY property: every caught-up non-leader's
    // vote/join position (`election_frontier`) equals its applied frontier in the
    // leader's stream (`applied_events`) — the same quantity in one numbering. A
    // bounded poll absorbs a brief marker-application lag (the frontier re-keys to
    // the new leader's shard as the marker folds), but a PERSISTENT cross-numbering
    // tear never clears and times out here.
    let cons_deadline = Instant::now() + Duration::from_secs(20);
    loop {
        let bad: Vec<ElectionSample> = (0..3)
            .filter_map(|i| election_sample(&cluster, i))
            .filter(|s| {
                !s.is_leader
                    && s.applied_events > 0
                    && s.lag_events == 0
                    && s.election_frontier != s.applied_events
            })
            .collect();
        if bad.is_empty() {
            break;
        }
        assert!(
            Instant::now() < cons_deadline,
            "[oracle] election-position INCONSISTENT at rest — a caught-up follower's \
             vote/join frontier disagrees with its applied frontier in the leader's \
             stream (the cross-numbering tear). Evidence: {bad:#?}"
        );
        std::thread::sleep(Duration::from_millis(200));
    }

    // Post-failover steady state: no node latched reseed, the corpus still serves.
    for idx in 0..3 {
        await_status_bool(
            &cluster,
            idx,
            "reseed_required",
            false,
            convergence_budget(),
            "the failover must not latch reseed on any node",
        );
    }
    println!("[oracle] consistency held across the failover — no tear, no false quarantine");
}