//! 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";

/// 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 (m12p6): the graceful-shutdown compaction now RETAINS the
/// `WAL_RETENTION_SEGMENTS` (= 4) 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` segments. At ~56 KiB/item the 16 MiB segment holds
/// ~292 items, so 1800 items ≈ 6 segments ⇒ the follower's frontier segment is
/// well past the 4-segment retention window and is genuinely deleted.
const OFFLINE_ITEMS: u64 = 1800;

/// 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");
}

/// 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)))
}

/// 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());
}

/// 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");

    // ── 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");
}

/// 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",
    );
    assert!(
        status_bool(&cluster, LEADER, "reseed_required"),
        "the quarantine latch must also write the reseed marker (m11p5 §2.4): {:?}",
        cluster.local_status(LEADER)
    );
    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");
}