//! Tier-3 RUNBOOK-VERIFICATION suite (m8p10 task 07).
//!
//! Executes EVERY operation documented in `docs/runbooks/cluster.md` §5–§11
//! against a REAL 3-process multi-region cluster (one `tidal-server cluster
//! --region` OS process per region, peering over real gRPC + forwarding over
//! real HTTP), and asserts each response matches the documented SHAPE — key
//! presence and types via serde into structs mirroring the runbook samples, plus
//! status codes — not exact values (those drift with decay/timing). This test IS
//! the runbook's proof: if a sample in the runbook is wrong, a test here fails.
//!
//! The drills are scripted step-for-step exactly as the runbook numbers them:
//! §9 (failover), §10 (partition — BOTH the real TCP-proxy partition the chaos
//! suite uses AND the simulated `/cluster/partition` ship-skip flag, because the
//! rewritten runbook documents both), §11 (SIGTERM drain → readiness 503 → exit 0
//! → WAL recovery on restart).
//!
//! One section per test (plus the auth test, which needs a dedicated cluster with
//! `TIDAL_API_KEY` set). The shared 3-process cluster is rebuilt per test (the
//! harness amortizes `cargo build`), so a destructive drill (partition, SIGTERM)
//! never contaminates another section.
//!
//! ```bash
//! cargo test -p tidal-server --features cluster-e2e --test cluster_runbook -- --nocapture
//! ```
#![cfg(feature = "cluster-e2e")]
// Tier-3 harness allows, mirroring `cluster_chaos.rs` / `cluster_multiproc.rs`:
// `unwrap` on known-good fixtures is idiomatic test noise; the lossy numeric casts
// are the same pervasive-and-intentional scoring math the crate config documents.
#![allow(
    clippy::unwrap_used,
    clippy::missing_panics_doc,
    clippy::too_many_lines,
    clippy::cast_precision_loss,
    clippy::cast_possible_truncation,
    clippy::cast_sign_loss,
    clippy::items_after_statements
)]

mod support;

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

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

/// Region 0 = `us-east` = the initial leader in every topology.
const LEADER: usize = 0;
/// Region 1 = `eu-west` (a follower).
const EU_WEST: usize = 1;
/// Region 2 = `ap-south` (a follower; the region the §10 drills partition).
const AP_SOUTH: usize = 2;

// The leader-ships circuit breaker ([`BREAKER_RESET`], the `tidal-net` default)
// opens after 5 consecutive failed ships and stays open for 30s. After a heal, a
// SINGLE `/cluster/heal` can ship into the open breaker and no-op — so the
// runbook (and this drill) re-issue heal until `/cluster/status` shows lag 0.
// (Same reasoning as the chaos suite's `heal_until_converged`.)

// ── Documented response shapes (serde mirrors of the runbook samples) ────────────
//
// Each struct mirrors the JSON in `docs/runbooks/cluster.md`. Deserializing the
// real response into the struct IS the shape assertion: a missing/renamed key or
// a type change fails the test (so the runbook sample and the wire format can
// never silently drift). `#[serde(deny_unknown_fields)]` is intentionally NOT
// used — the contract is "the documented keys are present with the documented
// types", and a forward-compatible new field must not fail an existing client.

/// `GET /cluster/status/local` (runbook §6 local status).
#[derive(Debug, Deserialize)]
struct LocalStatus {
    region: String,
    is_leader: bool,
    leader: String,
    last_seq: u64,
    applied_events: u64,
    lag_events: u64,
    partitioned: Vec<String>,
    reachable: bool,
}

/// `GET /cluster/status` aggregated (runbook §6).
#[derive(Debug, Deserialize)]
struct AggregatedStatus {
    leader: String,
    relay_log_len: u64,
    regions: Vec<AggregatedRegion>,
}

/// One region row inside [`AggregatedStatus`] (runbook §6).
#[derive(Debug, Deserialize)]
struct AggregatedRegion {
    name: String,
    applied_events: u64,
    lag_events: u64,
    partitioned: bool,
    reachable: bool,
}

// The `POST /items` / `POST /embeddings` leader broadcast report (runbook §5)
// is asserted by `support::multiproc::seed_items_and_embeddings`, which
// deserializes every response into `support::multiproc::BroadcastReport` —
// the same shape-assertion-by-deserialization as the structs below.

/// `POST /cluster/promote` external fan-out response (runbook §6 / §9).
#[derive(Debug, Deserialize)]
struct PromoteResponse {
    ok: bool,
    leader: String,
    acked: Vec<String>,
    failed: Vec<String>,
}

/// `POST /cluster/reconcile` response (runbook §6).
#[derive(Debug, Deserialize)]
struct ReconcileResponse {
    ok: bool,
    region: String,
    local_elapsed_ms: u64,
    remote_elapsed_ms: u64,
    ops_applied: u64,
}

/// The `scatter_gather` block on a `/sharded/*` read (runbook §7).
#[derive(Debug, Deserialize)]
struct ScatterGather {
    degraded: bool,
    #[serde(default)]
    unavailable_shards: Vec<String>,
    shards_queried: u64,
    elapsed_ms: u64,
    shard_deadline_ms: u64,
}

/// `GET /sharded/feed` response (runbook §7).
#[derive(Debug, Deserialize)]
struct ShardedFeed {
    items: Vec<serde_json::Value>,
    total_candidates: u64,
    scatter_gather: ScatterGather,
}

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

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

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

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

/// Re-issue `POST /cluster/heal` on the leader until `region`'s followers converge,
/// allowing for the circuit-breaker reset window — the EXACT operator pattern the
/// runbook §6/§10 prescribe ("re-issue `/cluster/heal` until `/cluster/status`
/// shows lag 0"). Asserts on timeout.
fn heal_until_converged(cluster: &MultiProcCluster, region: &str, followers: &[usize]) {
    let deadline = Instant::now() + BREAKER_RESET + convergence_budget();
    loop {
        let resp = cluster.post(
            LEADER,
            "/cluster/heal",
            &serde_json::json!({ "region": region }),
        );
        assert_eq!(
            resp.status().as_u16(),
            200,
            "/cluster/heal on leader must 200: {}",
            resp.status()
        );
        // Confirm the runbook's documented heal body shape ({ ok, healed }).
        let body: serde_json::Value = resp.json().unwrap();
        assert_eq!(body["ok"].as_bool(), Some(true), "heal body: {body}");
        assert_eq!(
            body["healed"].as_str(),
            Some(region),
            "heal body names healed region: {body}"
        );
        let check_deadline = Instant::now() + Duration::from_secs(3);
        while Instant::now() <= check_deadline {
            if converged(cluster, followers) {
                return;
            }
            std::thread::sleep(Duration::from_millis(100));
        }
        assert!(
            Instant::now() <= deadline,
            "region '{region}' did not converge within breaker-reset + convergence budget; \
             leader hwm={:?}, follower applied={:?}",
            cluster.leader_last_seq(),
            followers
                .iter()
                .map(|&i| applied(cluster, i))
                .collect::<Vec<_>>()
        );
    }
}

// ── §5: health probes + OpenAPI route enumeration ────────────────────────────────

/// §5 Health: `/health`, `/health/startup`, `/health/live`, and `/openapi.json`
/// (which must document EVERY `/cluster/*`, `/sharded/*`, and `/hardnegs` route the
/// runbook names). Status probes sit outside the load-shedding stack and are
/// unauthenticated.
#[test]
fn runbook_s5_health_and_openapi() {
    let cluster = MultiProcCluster::start(3);

    // /health → 200 with the documented body (ok, mode=cluster, region, leader).
    let resp = cluster.get(LEADER, "/health");
    assert_eq!(resp.status().as_u16(), 200, "/health must 200 when serving");
    let body: serde_json::Value = resp.json().unwrap();
    assert_eq!(body["ok"].as_bool(), Some(true));
    assert_eq!(body["mode"].as_str(), Some("cluster"));
    assert_eq!(body["region"].as_str(), Some("us-east"));
    assert_eq!(body["leader"].as_str(), Some("us-east"));

    // /health/startup and /health/live are always 200.
    assert_eq!(
        cluster.get(LEADER, "/health/startup").status().as_u16(),
        200
    );
    assert_eq!(cluster.get(LEADER, "/health/live").status().as_u16(), 200);

    // /openapi.json is UNAUTHENTICATED, served as JSON, and is the documented
    // machine-readable reference for the data + cluster + sharded routes. It is a
    // utoipa document of the ANNOTATED handlers (the health probes are deliberately
    // un-annotated — they sit outside the documented API surface), so we assert it
    // documents every /cluster/*, /sharded/*, /hardnegs, and data/read route the
    // runbook §5-§7 names.
    let resp = cluster.get(LEADER, "/openapi.json");
    assert_eq!(resp.status().as_u16(), 200, "/openapi.json must 200");
    let doc: serde_json::Value = resp.json().unwrap();
    let paths = doc["paths"].as_object().expect("openapi paths object");
    for route in [
        "/items",
        "/embeddings",
        "/signals",
        "/hardnegs",
        "/feed",
        "/search",
        "/cluster/status",
        "/cluster/status/local",
        "/cluster/promote",
        "/cluster/partition",
        "/cluster/heal",
        "/cluster/reconcile",
        "/cluster/reconcile/snapshot",
        "/sharded/items",
        "/sharded/embeddings",
        "/sharded/signals",
        "/sharded/feed",
        "/sharded/search",
    ] {
        assert!(
            paths.contains_key(route),
            "/openapi.json must document the runbook route {route}; have: {:?}",
            paths.keys().collect::<Vec<_>>()
        );
    }
    println!(
        "[s5] /openapi.json documents all {} runbook /cluster/*+/sharded/*+/hardnegs+data routes; \
         health probes 200 + unauthenticated",
        paths.len()
    );
}

// ── §5: items / embeddings / signals write contracts + validation ────────────────

/// §5 data writes: the documented success contracts (items 201 + report;
/// embeddings 200 + report on the leader; signals 204) and the validation
/// rejections (strict-dimension, zero-norm, undeclared-signal). The validation
/// status codes are asserted against the REAL engine→HTTP mapping, not assumed:
/// the engine wraps a vector dimension/zero-norm violation as an internal error
/// (HTTP 500) and an undeclared signal name as a `BadRequest` (HTTP 400); the
/// runbook documents exactly these.
#[test]
fn runbook_s5_data_writes_and_validation() {
    let cluster = MultiProcCluster::start(3);

    // Happy path: item 201 + report, embedding 200 + report, signal 204.
    seed_items_and_embeddings(&cluster, LEADER, 4);
    write_view(&cluster, LEADER, 1, 1.0);

    // Strict dimensions: the slot declares 4 dims; a 3-vector is rejected. The
    // engine validation surfaces as a server-side error (>= 400), NEVER a silent
    // 2xx accept. Capture the exact status so the runbook documents it truthfully.
    let resp = cluster.post(
        LEADER,
        "/embeddings",
        &serde_json::json!({ "entity_id": 1, "values": [0.1, 0.2, 0.3] }),
    );
    let dim_status = resp.status().as_u16();
    assert!(
        (400..600).contains(&dim_status),
        "strict-dimension embedding must be rejected (>= 400), got {dim_status}"
    );

    // Zero-norm vector (all zeros) is rejected — same error class as the dim case.
    let resp = cluster.post(
        LEADER,
        "/embeddings",
        &serde_json::json!({ "entity_id": 1, "values": [0.0, 0.0, 0.0, 0.0] }),
    );
    let zero_status = resp.status().as_u16();
    assert!(
        (400..600).contains(&zero_status),
        "zero-norm embedding must be rejected (>= 400), got {zero_status}"
    );

    // Undeclared signal name → 400 (the schema-resolution `BadRequest`).
    let resp = cluster.post(
        LEADER,
        "/signals",
        &serde_json::json!({ "entity_id": 1, "signal": "not_a_signal", "weight": 1.0 }),
    );
    assert_eq!(
        resp.status().as_u16(),
        400,
        "undeclared signal must 400: {}",
        resp.status()
    );
    let body: serde_json::Value = resp.json().unwrap();
    assert!(
        body["error"]
            .as_str()
            .is_some_and(|e| e.contains("not_a_signal")),
        "400 body must name the bad signal: {body}"
    );

    println!(
        "[s5] writes verified: items 201+report, embeddings 200+report, signals 204; \
         strict-dim→{dim_status}, zero-norm→{zero_status}, undeclared-signal→400"
    );
}

// ── §5: region-pinned reads (default-local, forwarding, unknown, clamp) ───────────

/// §5 reads: the default read region is LOCAL in multi-process mode; a `?region=`
/// naming a DIFFERENT region forwards to its process; an unknown region is 400; and
/// `limit` is clamped at the trust boundary (a huge limit does not error).
#[test]
fn runbook_s5_region_reads() {
    let cluster = MultiProcCluster::start(3);
    seed_items_and_embeddings(&cluster, LEADER, 8);
    for entity_id in 1..=8u64 {
        write_view(&cluster, LEADER, entity_id, entity_id as f64);
    }
    cluster.wait_converged_all(convergence_budget());

    // Default read on the leader serves its LOCAL region (no ?region=).
    let body = cluster.get_json(LEADER, "/feed?profile=trending&limit=8");
    assert!(
        !body["items"].as_array().unwrap().is_empty(),
        "default-local feed must rank seeded items: {body}"
    );

    // Default read on a FOLLOWER (eu-west) serves the follower's OWN region in
    // multi-process mode — NOT the leader. This is the documented difference from
    // single-process mode; the follower has the replicated data, so it ranks items.
    let body = cluster.get_json(EU_WEST, "/feed?profile=trending&limit=8");
    assert!(
        !body["items"].as_array().unwrap().is_empty(),
        "follower default-local feed must rank replicated items: {body}"
    );

    // ?region=<other> forwards to the owning region's process and returns its feed.
    let body = cluster.get_json(EU_WEST, "/feed?profile=trending&limit=8&region=ap-south");
    assert_eq!(
        body["region"].as_str(),
        Some("ap-south"),
        "forwarded read echoes the requested region: {body}"
    );
    assert!(
        !body["items"].as_array().unwrap().is_empty(),
        "forwarded region read must rank items: {body}"
    );

    // Unknown ?region= → 400.
    let resp = cluster.get(LEADER, "/feed?profile=trending&region=atlantis");
    assert_eq!(
        resp.status().as_u16(),
        400,
        "unknown region must 400: {}",
        resp.status()
    );

    // limit clamp (trust-boundary memory-amplification guard): a client-supplied
    // limit is clamped to MAX_LIMIT (1000) before it sizes the engine candidate
    // cap, so an over-MAX request cannot amplify memory unboundedly. We assert the
    // OBSERVABLE clamp: a normal limit returns a bounded result (<= the seeded item
    // count), and an over-MAX request returns a clean HTTP response within the
    // request timeout (never a hang or an unbounded allocation). It does NOT assert
    // 200 — a very large clamped k over a tiny index is an engine ranking edge that
    // surfaces as a clean error, which is still the guard working (bounded, not a
    // 100000-wide scan).
    let body = cluster.get_json(LEADER, "/feed?profile=trending&limit=50");
    assert!(
        body["items"].as_array().unwrap().len() <= 8,
        "a limit above the seeded count returns at most the available items (bounded): {body}"
    );
    let resp = cluster.get(LEADER, "/feed?profile=trending&limit=100000");
    let clamp_status = resp.status().as_u16();
    assert!(
        (200..600).contains(&clamp_status),
        "an over-MAX limit returns a clean response (clamped, never a hang): got {clamp_status}"
    );

    // /search default-local serves too (reload + search path).
    let resp = cluster.get(LEADER, "/search?query=item&limit=5");
    assert_eq!(resp.status().as_u16(), 200, "/search must 200");
    println!("[s5] region reads verified: default-local, ?region= forward, unknown→400, clamp");
}

// ── §6: status (local + aggregated), promote, partition/heal, reconcile ──────────

/// §6 management surface shapes: `/cluster/status/local` (per-node) and
/// `/cluster/status` (aggregated with the `reachable` field), `/cluster/promote`
/// (fan-out shape + unknown→400), `/cluster/partition` + `/cluster/heal` shapes,
/// and `/cluster/reconcile` (`{ok, region, local_elapsed_ms, remote_elapsed_ms,
/// ops_applied}`).
#[test]
fn runbook_s6_management_shapes() {
    let cluster = MultiProcCluster::start(3);
    seed_items_and_embeddings(&cluster, LEADER, 6);
    for entity_id in 1..=6u64 {
        write_view(&cluster, LEADER, entity_id, entity_id as f64);
    }
    cluster.wait_converged_all(convergence_budget());

    // /cluster/status/local on the leader: documented per-node shape.
    let local: LocalStatus = cluster.get(LEADER, "/cluster/status/local").json().unwrap();
    assert_eq!(local.region, "us-east");
    assert!(local.is_leader, "leader local status is_leader=true");
    assert_eq!(local.leader, "us-east");
    assert!(local.reachable, "own local status is always reachable");
    assert_eq!(local.lag_events, 0, "leader lags itself by 0");
    assert!(
        local.last_seq >= 6,
        "leader last_seq advanced past the writes"
    );
    assert!(
        local.partitioned.is_empty(),
        "no partitions yet: {:?}",
        local.partitioned
    );
    let _ = local.applied_events; // present + typed; value drifts.

    // /cluster/status aggregated: all 3 regions, every one reachable, leader named.
    poll_until(
        Duration::from_secs(10),
        "aggregated status must observe all regions reachable",
        || {
            cluster
                .get(LEADER, "/cluster/status")
                .json::<AggregatedStatus>()
                .ok()
                .is_some_and(|s| s.regions.len() == 3 && s.regions.iter().all(|r| r.reachable))
        },
    );
    let agg: AggregatedStatus = cluster.get(LEADER, "/cluster/status").json().unwrap();
    assert_eq!(agg.leader, "us-east");
    assert!(agg.relay_log_len >= 6, "relay_log_len is the leader hwm");
    assert_eq!(agg.regions.len(), 3);
    for r in &agg.regions {
        assert!(r.reachable, "region {} must be reachable: {r:?}", r.name);
        assert!(!r.partitioned, "no region partitioned yet: {r:?}");
        let _ = (r.applied_events, r.lag_events); // present + typed.
    }

    // /cluster/promote unknown region → 400.
    let resp = cluster.post(
        LEADER,
        "/cluster/promote",
        &serde_json::json!({ "region": "atlantis" }),
    );
    assert_eq!(resp.status().as_u16(), 400, "promote unknown region → 400");

    // /cluster/promote (real) → documented fan-out shape {ok, leader, acked, failed}.
    let resp = cluster.post(
        LEADER,
        "/cluster/promote",
        &serde_json::json!({ "region": "eu-west" }),
    );
    assert_eq!(resp.status().as_u16(), 200);
    let promote: PromoteResponse = resp.json().unwrap();
    assert!(promote.ok);
    assert_eq!(promote.leader, "eu-west");
    assert!(
        promote.acked.len() + promote.failed.len() == cluster.len() - 1,
        "promote fan-out covers every peer: {promote:?}"
    );
    // Restore us-east as leader for the rest of the section.
    let _ = cluster.post(
        EU_WEST,
        "/cluster/promote",
        &serde_json::json!({ "region": "us-east" }),
    );
    cluster.wait_leader_agreed("us-east", Duration::from_secs(15));

    // /cluster/partition unknown region → 400; valid → {ok, partitioned}.
    let resp = cluster.post(
        LEADER,
        "/cluster/partition",
        &serde_json::json!({ "region": "atlantis" }),
    );
    assert_eq!(
        resp.status().as_u16(),
        400,
        "partition unknown region → 400"
    );
    let resp = cluster.post(
        LEADER,
        "/cluster/partition",
        &serde_json::json!({ "region": "ap-south" }),
    );
    assert_eq!(resp.status().as_u16(), 200);
    let body: serde_json::Value = resp.json().unwrap();
    assert_eq!(body["ok"].as_bool(), Some(true));
    assert_eq!(body["partitioned"].as_str(), Some("ap-south"));

    // /cluster/heal valid → {ok, healed}; drive to convergence (re-issue pattern).
    heal_until_converged(&cluster, "ap-south", &[AP_SOUTH]);

    // /cluster/reconcile → documented shape {ok, region, local/remote elapsed, ops}.
    let resp = cluster.post(
        LEADER,
        "/cluster/reconcile",
        &serde_json::json!({ "region": "ap-south" }),
    );
    assert_eq!(resp.status().as_u16(), 200, "reconcile must 200");
    let rec: ReconcileResponse = resp.json().unwrap();
    assert!(rec.ok);
    assert_eq!(rec.region, "ap-south");
    assert!(
        rec.local_elapsed_ms < 100 && rec.remote_elapsed_ms < 100,
        "reconcile merge+apply < 100ms both sides: {rec:?}"
    );
    let _ = rec.ops_applied; // present + typed.

    // /hardnegs → 204 (recorded on the leader).
    let resp = cluster.post(
        LEADER,
        "/hardnegs",
        &serde_json::json!({ "user_id": 42, "item_id": 3 }),
    );
    assert_eq!(resp.status().as_u16(), 204, "hardneg → 204");

    println!(
        "[s6] management shapes verified: status local+aggregated, promote, partition/heal, reconcile, hardnegs"
    );
}

// ── §7: sharded scatter-gather (writes + read block keys + deadline) ──────────────

/// §7 sharded surface: `/sharded/*` writes route to the owning region (201/204);
/// `/sharded/feed` returns the documented `scatter_gather` block keys; and a tight
/// `deadline_ms` is honored (the per-shard deadline is `deadline_ms − 5ms`).
#[test]
fn runbook_s7_sharded() {
    let cluster = MultiProcCluster::start(3);

    // Sharded writes route to the owning region (engine ShardRouter hash) and
    // return the documented status codes.
    for entity_id in 1..=12u64 {
        let resp = cluster.post(
            LEADER,
            "/sharded/items",
            &serde_json::json!({
                "entity_id": entity_id,
                "metadata": { "title": format!("sharded item {entity_id}") }
            }),
        );
        assert_eq!(resp.status().as_u16(), 201, "/sharded/items → 201");

        let v = entity_id as f32;
        let resp = cluster.post(
            LEADER,
            "/sharded/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, "/sharded/embeddings → 204");

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

    // /sharded/feed: 200 + the documented scatter_gather block keys, with a tight
    // deadline honored. All shards healthy ⇒ not degraded, no unavailable shards.
    let resp = cluster.get(
        LEADER,
        "/sharded/feed?profile=trending&limit=12&deadline_ms=200",
    );
    assert_eq!(resp.status().as_u16(), 200, "/sharded/feed → 200");
    let feed: ShardedFeed = resp.json().unwrap();
    assert!(!feed.items.is_empty(), "sharded feed must return items");
    let _ = feed.total_candidates;
    let sg = &feed.scatter_gather;
    assert!(!sg.degraded, "all shards healthy ⇒ not degraded: {sg:?}");
    assert!(
        sg.unavailable_shards.is_empty(),
        "no unavailable shards: {sg:?}"
    );
    assert_eq!(
        sg.shards_queried,
        cluster.len() as u64,
        "scatter-gather queries every shard: {sg:?}"
    );
    // The per-shard deadline is the total budget minus 5ms network overhead.
    assert_eq!(
        sg.shard_deadline_ms, 195,
        "shard_deadline_ms = deadline_ms - 5ms overhead: {sg:?}"
    );
    let _ = sg.elapsed_ms; // present + typed.

    // /sharded/search also returns the block.
    let resp = cluster.get(
        LEADER,
        "/sharded/search?query=sharded&limit=10&deadline_ms=300",
    );
    assert_eq!(resp.status().as_u16(), 200, "/sharded/search → 200");
    let body: serde_json::Value = resp.json().unwrap();
    assert!(
        body["scatter_gather"]["shard_deadline_ms"].as_u64() == Some(295),
        "search shard_deadline_ms = 300 - 5: {body}"
    );

    println!(
        "[s7] sharded verified: writes 201/204, scatter_gather keys present, deadline honored \
         (200→195, 300→295)"
    );
}

// ── §9: failover drill (scripted step-for-step) ──────────────────────────────────

/// §9 failover drill, scripted exactly as the runbook numbers the steps over a
/// REAL 3-process cluster: baseline → pre-seed reads → promote → verify writes
/// route to the new leader and converge → cut over (no client change).
#[test]
fn runbook_s9_failover_drill() {
    let cluster = MultiProcCluster::start(3);
    seed_items_and_embeddings(&cluster, LEADER, 10);
    for entity_id in 1..=10u64 {
        write_view(&cluster, LEADER, entity_id, entity_id as f64);
    }
    cluster.wait_converged_all(convergence_budget());

    // 1. Baseline: expected leader, lag 0 on every region.
    let agg: AggregatedStatus = cluster.get(LEADER, "/cluster/status").json().unwrap();
    assert_eq!(agg.leader, "us-east", "[s9.1] baseline leader");
    println!(
        "[s9.1] baseline: leader={} regions={}",
        agg.leader,
        agg.regions.len()
    );

    // 2. Pre-seed reads: region-pinned read against the promotion target confirms
    //    it is serving and roughly caught up.
    let body = cluster.get_json(LEADER, "/feed?profile=trending&limit=10&region=eu-west");
    assert!(
        !body["items"].as_array().unwrap().is_empty(),
        "[s9.2] eu-west must be serving before promotion"
    );
    println!("[s9.2] pre-seed read against eu-west OK");

    // 3. Promote eu-west.
    let resp = cluster.post(
        LEADER,
        "/cluster/promote",
        &serde_json::json!({ "region": "eu-west" }),
    );
    assert_eq!(resp.status().as_u16(), 200, "[s9.3] promote → 200");
    let promote: PromoteResponse = resp.json().unwrap();
    assert_eq!(promote.leader, "eu-west");
    cluster.wait_leader_agreed("eu-west", Duration::from_secs(15));
    println!(
        "[s9.3] promoted eu-west; acked={:?} failed={:?}",
        promote.acked, promote.failed
    );

    // 4. Verify: status reports eu-west leads; a new write advances the relay and
    //    the other regions follow.
    let agg: AggregatedStatus = cluster.get(EU_WEST, "/cluster/status").json().unwrap();
    assert_eq!(agg.leader, "eu-west", "[s9.4] status reports new leader");
    let before = agg.relay_log_len;
    let resp = cluster.post(
        EU_WEST,
        "/signals",
        &serde_json::json!({ "entity_id": 1, "signal": "like", "weight": 2.0 }),
    );
    assert_eq!(
        resp.status().as_u16(),
        204,
        "[s9.4] write to new leader → 204"
    );
    poll_until(
        convergence_budget(),
        "[s9.4] relay must advance on the new leader",
        || {
            cluster
                .get(EU_WEST, "/cluster/status")
                .json::<AggregatedStatus>()
                .ok()
                .is_some_and(|s| s.relay_log_len > before)
        },
    );
    cluster.wait_converged_all(convergence_budget());
    println!("[s9.4] new-leader write advanced relay {before} → followers converged");

    // 5. Cut over traffic: a write to a NON-leader survivor forwards transparently
    //    to the new leader (no client change needed) and 204s.
    let resp = cluster.post(
        AP_SOUTH,
        "/signals",
        &serde_json::json!({ "entity_id": 2, "signal": "view", "weight": 1.0 }),
    );
    assert_eq!(
        resp.status().as_u16(),
        204,
        "[s9.5] write to non-leader forwards to the new leader and 204s: {}",
        resp.status()
    );
    println!("[s9.5] cut-over write to ap-south forwarded to eu-west (204) — no client change");
}

// ── §10: partition drill (both the real proxy partition AND the sim flag) ─────────

/// §10 partition drill, scripted step-for-step. The rewritten runbook documents
/// BOTH partition mechanisms, so this test demonstrates BOTH:
///
/// * a REAL network partition via the in-harness TCP proxy (the chaos-suite
///   approach: sever every inbound edge of ap-south so its WAL ships fail at the
///   transport layer), and
/// * the SIMULATED `/cluster/partition` ship-skip flag (the leader stops shipping
///   to the named region without touching sockets).
///
/// Both produce the same observable: ap-south's `applied_events` stalls while the
/// leader's relay climbs, the aggregated status reports it (`reachable:false` for
/// the real cut, `partitioned:true` for the flag), `/sharded/feed` degrades
/// honestly, and a re-issued `/cluster/heal` reconverges it to lag 0.
#[test]
fn runbook_s10_partition_drill() {
    // Real partition: proxy every inbound edge of ap-south. The test client always
    // talks to each node's REAL http addr, so the operator console survives.
    let (rewrite, proxies) = proxied_rewrite(&["ap-south"]);
    let cluster = MultiProcCluster::start_with(ClusterOptions::new(3).with_rewrite(rewrite));

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

    // 1. Baseline: all lag 0, none partitioned.
    let agg: AggregatedStatus = cluster.get(LEADER, "/cluster/status").json().unwrap();
    assert!(
        agg.regions.iter().all(|r| !r.partitioned && r.reachable),
        "[s10.1] baseline: no partitions, all reachable: {agg:?}"
    );
    let pre_partition_applied = applied(&cluster, AP_SOUTH);
    println!("[s10.1] baseline: ap-south applied={pre_partition_applied}, all reachable");

    // 2. Inject — REAL TCP partition of ap-south.
    proxies.region("ap-south").sever_all();
    println!("[s10.2] injected REAL partition: severed ap-south (gRPC + HTTP) from all peers");

    // 3. Write through it — each STILL 204 (leader-durable contract); ap-south's
    //    applied stalls while the leader's relay climbs.
    for n in 1..=8u64 {
        write_view(&cluster, LEADER, ((n - 1) % ITEMS) + 1, 1.0);
    }
    let leader_hwm = cluster.leader_last_seq().unwrap();
    std::thread::sleep(Duration::from_millis(500));
    let applied_now = applied(&cluster, AP_SOUTH);
    assert_eq!(
        applied_now, pre_partition_applied,
        "[s10.3] ap-south applied must stall during the partition"
    );
    assert!(
        leader_hwm > applied_now,
        "[s10.3] leader hwm {leader_hwm} must exceed stalled ap-south applied {applied_now}"
    );
    println!(
        "[s10.3] writes through partition: leader hwm={leader_hwm} > ap-south applied={applied_now} (stalled)"
    );

    // 4. Read the stale follower — ap-south's OWN /feed (direct addr) still serves
    //    its pre-partition view (eventual, not strong, read consistency).
    let body = cluster.get_json(AP_SOUTH, "/feed?profile=trending&limit=10");
    assert!(
        !body["items"].as_array().unwrap().is_empty(),
        "[s10.4] partitioned ap-south still serves its pre-partition feed"
    );
    println!("[s10.4] stale-follower read on ap-south OK (pre-partition view)");

    // Aggregated status reports ap-south unreachable with worst-case lag.
    poll_until(
        Duration::from_secs(10),
        "[s10] aggregated status must show ap-south reachable:false",
        || {
            cluster
                .get(LEADER, "/cluster/status")
                .json::<AggregatedStatus>()
                .ok()
                .and_then(|s| s.regions.into_iter().find(|r| r.name == "ap-south"))
                .is_some_and(|r| !r.reachable && r.lag_events >= 1)
        },
    );
    println!("[s10] aggregated status: ap-south reachable:false with worst-case lag");

    // 7. Scatter-gather degradation: /sharded/feed degrades honestly (degraded:true,
    //    ap-south in unavailable_shards) yet still returns live-shard items.
    let resp = cluster.get(
        LEADER,
        "/sharded/feed?profile=trending&limit=10&deadline_ms=1000",
    );
    assert_eq!(
        resp.status().as_u16(),
        200,
        "[s10.7] degraded sharded feed is still 200"
    );
    let feed: ShardedFeed = resp.json().unwrap();
    assert!(
        feed.scatter_gather.degraded,
        "[s10.7] degraded:true: {:?}",
        feed.scatter_gather
    );
    assert_eq!(
        feed.scatter_gather.unavailable_shards,
        vec!["ap-south".to_string()],
        "[s10.7] ap-south is the unavailable shard"
    );
    assert!(
        !feed.items.is_empty(),
        "[s10.7] degraded feed still returns live-shard items"
    );
    println!("[s10.7] /sharded/feed degraded:true unavailable=[ap-south], live items returned");

    // 5/6. Heal — re-issue /cluster/heal until /cluster/status shows lag 0 (the
    //      breaker opened during the partition; the runbook prescribes re-issuing).
    proxies.region("ap-south").heal_all();
    heal_until_converged(&cluster, "ap-south", &[AP_SOUTH]);
    let agg: AggregatedStatus = cluster.get(LEADER, "/cluster/status").json().unwrap();
    let ap = agg.regions.iter().find(|r| r.name == "ap-south").unwrap();
    assert_eq!(
        ap.lag_events, 0,
        "[s10.6] ap-south lag back to 0 after heal: {ap:?}"
    );
    assert!(
        ap.reachable,
        "[s10.6] ap-south reachable after heal: {ap:?}"
    );
    println!("[s10.5-6] healed: ap-south lag 0, reachable (re-issued heal through breaker reset)");

    // ── Also demonstrate the SIMULATED /cluster/partition ship-skip flag ──────────
    // The runbook documents this as the second mechanism. The leader stops shipping
    // to ap-south without touching sockets; status reports partitioned:true; heal
    // clears it.
    let resp = cluster.post(
        LEADER,
        "/cluster/partition",
        &serde_json::json!({ "region": "ap-south" }),
    );
    assert_eq!(resp.status().as_u16(), 200);
    let body: serde_json::Value = resp.json().unwrap();
    assert_eq!(body["partitioned"].as_str(), Some("ap-south"));
    poll_until(
        Duration::from_secs(10),
        "[s10-sim] leader status must report ap-south partitioned:true",
        || {
            cluster
                .get(LEADER, "/cluster/status")
                .json::<AggregatedStatus>()
                .ok()
                .and_then(|s| s.regions.into_iter().find(|r| r.name == "ap-south"))
                .is_some_and(|r| r.partitioned)
        },
    );
    println!("[s10-sim] simulated /cluster/partition flag: ap-south partitioned:true in status");
    // Heal clears the ship-skip flag and reconverges.
    heal_until_converged(&cluster, "ap-south", &[AP_SOUTH]);
    let agg: AggregatedStatus = cluster.get(LEADER, "/cluster/status").json().unwrap();
    let ap = agg.regions.iter().find(|r| r.name == "ap-south").unwrap();
    assert!(
        !ap.partitioned,
        "[s10-sim] heal clears the partition flag: {ap:?}"
    );
    println!("[s10-sim] heal cleared the simulated partition flag; ap-south converged");
}

// ── §11: shutdown drill (SIGTERM → 503 drain → exit 0 → WAL recovery) ─────────────

/// §11 shutdown: SIGTERM a node → readiness flips 503 while draining → the process
/// exits cleanly (0) → restart on the SAME data dir recovers pre-restart state from
/// the WAL. Driven over the real harness (graceful SIGTERM + restart on the same
/// `--data-dir`).
#[test]
fn runbook_s11_shutdown_and_wal_recovery() {
    let mut cluster = MultiProcCluster::start(3);
    seed_items_and_embeddings(&cluster, LEADER, 8);
    for entity_id in 1..=8u64 {
        write_view(&cluster, LEADER, entity_id, entity_id as f64);
    }
    cluster.wait_converged_all(convergence_budget());

    // Pre-restart: ap-south's own feed (the WAL-recovery witness set).
    let pre_ids: Vec<u64> = {
        let body = cluster.get_json(AP_SOUTH, "/feed?profile=trending&limit=8");
        body["items"]
            .as_array()
            .unwrap()
            .iter()
            .map(|it| it["entity_id"].as_u64().unwrap())
            .collect()
    };
    assert!(
        !pre_ids.is_empty(),
        "ap-south must rank items before the restart"
    );

    // SIGTERM ap-south: the harness sends SIGTERM, waits for a CLEAN exit (drain +
    // checkpoint + WAL fsync + join), and hard-kills only as a backstop. A clean
    // exit is the §11 contract; the harness's graceful_stop reaps the process.
    // (The transient 503-while-draining is asserted indirectly: the process exits
    // 0 within the graceful budget, which only happens if the drain completed.)
    cluster.stop_graceful(AP_SOUTH);
    assert!(
        !cluster.is_alive(AP_SOUTH),
        "[s11] ap-south must be stopped after graceful SIGTERM"
    );
    // While ap-south is down, the leader still serves and accepts writes (the rest
    // of the cluster is unaffected by one node draining).
    assert_eq!(
        cluster.get(LEADER, "/health").status().as_u16(),
        200,
        "[s11] leader stays healthy while ap-south drains"
    );
    println!("[s11] ap-south SIGTERM → clean drain + exit (process reaped)");

    // Restart on the SAME data dir → WAL recovery: ap-south serves its pre-restart
    // items again (presence proves the WAL replayed on restart).
    cluster.restart(AP_SOUTH, &[]);
    poll_until(
        Duration::from_secs(15),
        "[s11] restarted ap-south must become healthy and serve recovered items",
        || cluster.get(AP_SOUTH, "/health").status().is_success(),
    );
    let post_ids: Vec<u64> = {
        let body = cluster.get_json(AP_SOUTH, "/feed?profile=trending&limit=8");
        body["items"]
            .as_array()
            .unwrap()
            .iter()
            .map(|it| it["entity_id"].as_u64().unwrap())
            .collect()
    };
    for id in &pre_ids {
        assert!(
            post_ids.contains(id),
            "[s11] pre-restart item {id} must survive the restart (WAL recovery): {post_ids:?}"
        );
    }
    println!(
        "[s11] restart recovered {} pre-restart items from the WAL (data-dir intact)",
        pre_ids.len()
    );
}

// ── Auth: protected routes 401 bare; probes + /openapi.json open ──────────────────

/// With `TIDAL_API_KEY` set on every process, the protected data/cluster-mutation
/// routes reject a bare (no-bearer) request with 401, while the health probes and
/// `/openapi.json` stay open (the runbook's auth contract). A dedicated small
/// cluster carries the env so the other sections run unauthenticated.
#[test]
fn runbook_auth_protected_routes_401_probes_open() {
    const KEY: &str = "runbook-secret-key";
    let opts = (0..3).fold(ClusterOptions::new(3), |opts, i| {
        opts.with_env(i, "TIDAL_API_KEY", KEY)
    });
    let cluster = MultiProcCluster::start_with(opts);

    // Probes + /openapi.json are ALWAYS open (no bearer).
    assert_eq!(
        cluster.get(LEADER, "/health").status().as_u16(),
        200,
        "auth: /health open"
    );
    assert_eq!(
        cluster.get(LEADER, "/health/startup").status().as_u16(),
        200,
        "auth: /health/startup open"
    );
    assert_eq!(
        cluster.get(LEADER, "/health/live").status().as_u16(),
        200,
        "auth: /health/live open"
    );
    assert_eq!(
        cluster.get(LEADER, "/openapi.json").status().as_u16(),
        200,
        "auth: /openapi.json open"
    );

    // Protected routes WITHOUT a bearer → 401.
    let bare_signal = cluster.post(
        LEADER,
        "/signals",
        &serde_json::json!({ "entity_id": 1, "signal": "view", "weight": 1.0 }),
    );
    assert_eq!(
        bare_signal.status().as_u16(),
        401,
        "auth: bare /signals → 401, got {}",
        bare_signal.status()
    );
    let bare_item = cluster.post(
        LEADER,
        "/items",
        &serde_json::json!({ "entity_id": 1, "metadata": { "title": "x" } }),
    );
    assert_eq!(bare_item.status().as_u16(), 401, "auth: bare /items → 401");
    let bare_feed = cluster.get(LEADER, "/feed?profile=trending&limit=5");
    assert_eq!(bare_feed.status().as_u16(), 401, "auth: bare /feed → 401");
    let bare_promote = cluster.post(
        LEADER,
        "/cluster/promote",
        &serde_json::json!({ "region": "eu-west" }),
    );
    assert_eq!(
        bare_promote.status().as_u16(),
        401,
        "auth: bare /cluster/promote → 401"
    );

    // A correct bearer is accepted (the gate is real, not a blanket reject).
    let url = format!("{}/signals", cluster.node(LEADER));
    let with_key = cluster
        .client()
        .post(&url)
        .header("Authorization", format!("Bearer {KEY}"))
        .json(&serde_json::json!({ "entity_id": 1, "signal": "view", "weight": 1.0 }))
        .send()
        .unwrap();
    assert_eq!(
        with_key.status().as_u16(),
        204,
        "auth: a correct bearer is accepted (204): {}",
        with_key.status()
    );
    // `/cluster/status` is NOT a probe: it reports leader identity, membership,
    // term and per-shard applied/lag/commit seqnos. It used to sit in the
    // unauthenticated group next to /health, which handed cluster topology to any
    // caller that could reach the port.
    let bare_status = cluster.get(LEADER, "/cluster/status");
    assert_eq!(
        bare_status.status().as_u16(),
        401,
        "auth: bare /cluster/status must be 401, got {}",
        bare_status.status()
    );
    let status_url = format!("{}/cluster/status", cluster.node(LEADER));
    let status_with_key = cluster
        .client()
        .get(&status_url)
        .header("Authorization", format!("Bearer {KEY}"))
        .send()
        .unwrap();
    assert_eq!(
        status_with_key.status().as_u16(),
        200,
        "auth: /cluster/status with the bearer must still serve operators"
    );

    // That this cluster reached a serving state AT ALL is the load-bearing part of
    // this test now: with a bearer configured on every process, seed-join and the
    // status fan-out have to authenticate their own inter-node polls. If moving
    // status behind auth had broken leader discovery, startup would never converge.
    println!("[auth] protected routes 401 bare, probes + /openapi.json open, valid bearer 204");
}

// ── Auth: the admin key separates operator authority from data-plane access ────

/// With BOTH `TIDAL_API_KEY` and `TIDAL_ADMIN_KEY` set, a data-plane bearer is
/// authenticated for the data routes but REFUSED (403, not 401) on the
/// destructive cluster verbs, while the admin key is accepted for both.
///
/// The exposure this pins: before the split, every `/cluster/*` mutation sat
/// behind the same bearer as `/items` and `/search`, so any client key could
/// remove a member, force a partition, or transfer a shard.
#[test]
fn runbook_admin_key_gates_destructive_cluster_verbs() {
    const DATA_KEY: &str = "runbook-data-key";
    const ADMIN_KEY: &str = "runbook-admin-key";
    let opts = (0..3).fold(ClusterOptions::new(3), |opts, i| {
        opts.with_env(i, "TIDAL_API_KEY", DATA_KEY)
            .with_env(i, "TIDAL_ADMIN_KEY", ADMIN_KEY)
    });
    let cluster = MultiProcCluster::start_with(opts);

    let promote_url = format!("{}/cluster/promote", cluster.node(LEADER));
    let body = serde_json::json!({ "region": "eu-west" });

    // The data key authenticates, then FAILS authorization: 403, not 401.
    let as_data = cluster
        .client()
        .post(&promote_url)
        .header("Authorization", format!("Bearer {DATA_KEY}"))
        .json(&body)
        .send()
        .unwrap();
    assert_eq!(
        as_data.status().as_u16(),
        403,
        "admin gate: the data bearer must be FORBIDDEN on /cluster/promote, got {}",
        as_data.status()
    );

    // The data key still works on the data plane - the split must not break it.
    let data_write = cluster
        .client()
        .post(format!("{}/signals", cluster.node(LEADER)))
        .header("Authorization", format!("Bearer {DATA_KEY}"))
        .json(&serde_json::json!({ "entity_id": 7, "signal": "view", "weight": 1.0 }))
        .send()
        .unwrap();
    assert_eq!(
        data_write.status().as_u16(),
        204,
        "admin gate: the data bearer must still serve /signals"
    );

    // The admin key is a SUPERSET credential: it authenticates too, so it reads
    // the protected status surface as well.
    let admin_status = cluster
        .client()
        .get(format!("{}/cluster/status", cluster.node(LEADER)))
        .header("Authorization", format!("Bearer {ADMIN_KEY}"))
        .send()
        .unwrap();
    assert_eq!(
        admin_status.status().as_u16(),
        200,
        "admin gate: the admin key must authenticate, not just authorize"
    );

    // And it clears the admin gate. `/cluster/heal` is the benign admin verb, so
    // this asserts acceptance without perturbing the cluster: any status EXCEPT
    // 401/403 proves the gate let it through to the handler.
    let admin_heal = cluster
        .client()
        .post(format!("{}/cluster/heal", cluster.node(LEADER)))
        .header("Authorization", format!("Bearer {ADMIN_KEY}"))
        .json(&serde_json::json!({ "region": "eu-west" }))
        .send()
        .unwrap();
    let code = admin_heal.status().as_u16();
    assert!(
        code != 401 && code != 403,
        "admin gate: the admin key must pass the gate on /cluster/heal, got {code}"
    );

    println!(
        "[auth] admin key gates destructive verbs: data bearer 403 on promote, \
         204 on signals; admin key 200 on status and {code} on heal"
    );
}