//! m8p10 in-process multi-process-cluster tests.
//!
//! Each test builds TWO `ShardReplica`s in ONE test process — distinct
//! topologies pointing at each other's REAL loopback gRPC addresses — and drives
//! them over real HTTP (axum on loopback) + real `GrpcTransport` replication.
//! Unlike `cluster_e2e.rs` (tier-3, spawns OS processes), these run in the
//! default test build with no OS processes, exactly like `cluster_grpc.rs`.
//!
//! They prove the multi-process region node: convergence over real loopback
//! gRPC with decay parity, the typed `NotLeader` rejection, partition→heal with
//! idempotent re-heal, and that promote flips roles with the ALWAYS-ON receiver
//! (the demoted node applies the new leader's ships).
#![allow(
    clippy::unwrap_used,
    clippy::missing_panics_doc,
    clippy::too_many_lines,
    clippy::doc_markdown
)]

use std::{
    net::{SocketAddr, TcpListener},
    sync::Arc,
    time::{Duration, Instant},
};

use tidal_server::cluster::{
    ClusterNode, ElectionSpec, RegionSpec, ReplicationSpec, ShardReplicaSpec, ShardSpec,
    TimeoutsSpec, TopologySpec, WalSpec, build_region_router,
};
use tidaldb::replication::shard::ShardRouter;
use tidaldb::schema::{DecaySpec, EntityId, EntityKind, Schema, SchemaBuilder, Window};
use tidaldb::wal::format::{MemberEntry, MemberRole, MembershipRecord};

/// Unauthenticated reloadable creds for the in-process router tests (no bearer,
/// no cluster key — the pre-m11p7 `api_key: None` posture).
fn mk_test_creds() -> Arc<tidal_server::cluster::security::ClusterCreds> {
    Arc::new(tidal_server::cluster::security::ClusterCreds::unauthenticated())
}

/// A single-`view`-signal schema with a `hide` hard-negative signal (so the
/// `/hardnegs` route's `signal_with_context("hide", …)` resolves).
fn region_schema() -> Schema {
    let mut builder = SchemaBuilder::new();
    let _ = builder
        .signal(
            "view",
            EntityKind::Item,
            DecaySpec::Exponential {
                half_life: Duration::from_secs(7 * 24 * 3600),
            },
        )
        .windows(&[Window::OneHour])
        .velocity(false)
        .add();
    let _ = builder
        .signal("hide", EntityKind::Item, DecaySpec::Permanent)
        .velocity(false)
        .add();
    builder.build().unwrap()
}

/// The `region_schema` plus an 8-dim Item `content` embedding slot, so the
/// `/vector_search` recall probe (m12p1) resolves a slot and the replicated
/// embeddings get indexed on the follower.
fn region_schema_emb() -> Schema {
    let mut builder = SchemaBuilder::new();
    let _ = builder
        .signal(
            "view",
            EntityKind::Item,
            DecaySpec::Exponential {
                half_life: Duration::from_secs(7 * 24 * 3600),
            },
        )
        .windows(&[Window::OneHour])
        .velocity(false)
        .add();
    builder.embedding_slot("content", EntityKind::Item, 8);
    builder.build().unwrap()
}

/// Build a region node whose schema carries the `content` embedding slot. Mirrors
/// [`build_region`] (off-reactor gRPC construction) with the embeddings schema.
fn build_region_emb(topology: TopologySpec, region: &str, dir: &tempfile::TempDir) -> ClusterNode {
    let region = region.to_string();
    let data_dir = dir.path().to_path_buf();
    std::thread::spawn(move || {
        ClusterNode::new(
            &topology,
            &region,
            region_schema_emb(),
            Vec::new(),
            Some(data_dir),
            0,
        )
    })
    .join()
    .unwrap()
    .expect("region node builds with real gRPC transport")
}

/// Reserve a free loopback port and return its address.
fn free_addr() -> SocketAddr {
    TcpListener::bind("127.0.0.1:0")
        .unwrap()
        .local_addr()
        .unwrap()
}

/// A pair of fully-declared (grpc + http) region specs that point at each
/// other. `leader` is the first region's name.
struct Pair {
    leader_name: String,
    follower_name: String,
    leader_grpc: SocketAddr,
    follower_grpc: SocketAddr,
    leader_http: SocketAddr,
    follower_http: SocketAddr,
}

impl Pair {
    fn new() -> Self {
        Self {
            leader_name: "us-east".into(),
            follower_name: "eu-west".into(),
            leader_grpc: free_addr(),
            follower_grpc: free_addr(),
            leader_http: free_addr(),
            follower_http: free_addr(),
        }
    }

    /// The shared topology both processes parse (same declaration order ⇒ same
    /// RegionIds in both). HTTP addrs are the in-test axum binds.
    fn topology(&self) -> TopologySpec {
        TopologySpec {
            regions: vec![
                RegionSpec {
                    name: self.leader_name.clone(),
                    grpc_addr: Some(self.leader_grpc.to_string()),
                    grpc_bind: None,
                    http_addr: Some(self.leader_http.to_string()),
                    grpc_tls: None,
                    metrics_addr: None,
                    zone: None,
                },
                RegionSpec {
                    name: self.follower_name.clone(),
                    grpc_addr: Some(self.follower_grpc.to_string()),
                    grpc_bind: None,
                    http_addr: Some(self.follower_http.to_string()),
                    grpc_tls: None,
                    metrics_addr: None,
                    zone: None,
                },
            ],
            leader: self.leader_name.clone(),
            write_workers: None,
            timeouts: TimeoutsSpec::default(),
            replication: ReplicationSpec::default(),
            wal: WalSpec::default(),
            election: ElectionSpec::default(),
            shards: None,
        }
    }
}

/// A node's persistent data dir (m11p2: the durable WAL is the replication
/// stream, so multi-process cluster mode requires one).
///
/// DECLARE THE DIR BEFORE ANYTHING THAT CAN HOLD THE NODE (including the
/// tokio runtime): locals — and bindings within one tuple pattern — drop in
/// reverse declaration order, and a dir deleted while its node still runs
/// wedges fjall's flush worker on `NotFound` (the sealed memtable then never
/// drains and `rotate_memtable_and_wait` polls forever — observed as a
/// permanently hung test on the panic-unwind path).
fn region_dir() -> tempfile::TempDir {
    tempfile::tempdir().expect("create per-region data dir")
}

/// Build one region node off the reactor (GrpcTransport::new blocks on its own
/// runtime, so it must run on a plain thread). `dir` is the node's data dir;
/// see [`region_dir`] for the declaration-order contract.
fn build_region(topology: TopologySpec, region: &str, dir: &tempfile::TempDir) -> ClusterNode {
    let region = region.to_string();
    let data_dir = dir.path().to_path_buf();
    std::thread::spawn(move || {
        ClusterNode::new(
            &topology,
            &region,
            region_schema(),
            Vec::new(),
            Some(data_dir),
            0,
        )
    })
    .join()
    .unwrap()
    .expect("region node builds with real gRPC transport")
}

/// Serve `router` on `addr` using `rt`; returns once the listener is bound.
fn serve(rt: &tokio::runtime::Runtime, router: axum::Router, addr: SocketAddr) {
    let listener = rt
        .block_on(tokio::net::TcpListener::bind(addr))
        .unwrap_or_else(|e| panic!("bind {addr}: {e}"));
    rt.spawn(async move {
        let _ = axum::serve(listener, router).await;
    });
}

/// Poll `GET /cluster/status/local` on `base` until `pred(applied, lag)` holds
/// or the deadline elapses.
fn poll_status(
    client: &reqwest::blocking::Client,
    base: &str,
    pred: impl Fn(u64, u64) -> bool,
) -> serde_json::Value {
    let deadline = Instant::now() + Duration::from_secs(5);
    loop {
        let status: serde_json::Value = client
            .get(format!("{base}/cluster/status/local"))
            .send()
            .unwrap()
            .json()
            .unwrap();
        let applied = status["applied_events"].as_u64().unwrap_or(0);
        let lag = status["lag_events"].as_u64().unwrap_or(u64::MAX);
        if pred(applied, lag) {
            return status;
        }
        assert!(
            Instant::now() <= deadline,
            "status predicate not met within 5s: {status}"
        );
        std::thread::sleep(Duration::from_millis(20));
    }
}

/// Read entity `entity`'s trending feed score on `base` (0.0 if absent).
fn feed_score(client: &reqwest::blocking::Client, base: &str, entity: u64) -> f64 {
    let feed: serde_json::Value = client
        .get(format!("{base}/feed?profile=trending&limit=10"))
        .send()
        .unwrap()
        .json()
        .unwrap();
    feed["items"]
        .as_array()
        .unwrap()
        .iter()
        .find(|it| it["entity_id"].as_u64() == Some(entity))
        .and_then(|it| it["score"].as_f64())
        .unwrap_or(0.0)
}

/// Two region nodes converge over real loopback gRPC; the follower's feed scores
/// match the leader's to 1e-6 (decay parity).
#[test]
fn region_node_replicates_over_grpc() {
    let pair = Pair::new();

    let leader_dir = region_dir();
    let leader = build_region(pair.topology(), &pair.leader_name, &leader_dir);
    let follower_dir = region_dir();
    let follower = build_region(pair.topology(), &pair.follower_name, &follower_dir);

    // Build the runtime AFTER the nodes (and their data-dir guards):
    // locals drop in reverse order, so the runtime — which owns the
    // nodes via the serve tasks — tears down BEFORE the dirs delete,
    // on the panic-unwind path too (m11p2: nodes are persistent now).
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .unwrap();
    serve(
        &rt,
        build_region_router(Arc::new(leader), mk_test_creds()),
        pair.leader_http,
    );
    serve(
        &rt,
        build_region_router(Arc::new(follower), mk_test_creds()),
        pair.follower_http,
    );

    let client = reqwest::blocking::Client::new();
    let leader_base = format!("http://{}", pair.leader_http);
    let follower_base = format!("http://{}", pair.follower_http);

    // Broadcast items to BOTH nodes (items are not WAL-replicated in this task),
    // then write signals on the leader (replicated to the follower over gRPC).
    for i in 1..=8u64 {
        for base in [&leader_base, &follower_base] {
            let resp = client
                .post(format!("{base}/items"))
                .json(&serde_json::json!({
                    "entity_id": i,
                    "metadata": { "title": format!("item {i}") }
                }))
                .send()
                .unwrap();
            assert!(resp.status().is_success(), "POST /items: {}", resp.status());
        }
        let resp = client
            .post(format!("{leader_base}/signals"))
            .json(&serde_json::json!({ "entity_id": i, "signal": "view", "weight": 1.0 }))
            .send()
            .unwrap();
        assert!(
            resp.status().is_success(),
            "POST /signals on leader: {}",
            resp.status()
        );
    }

    // Follower converges: applied reaches 8 and lag returns to 0.
    poll_status(&client, &follower_base, |applied, lag| {
        applied >= 8 && lag == 0
    });

    // Decay parity: the leader and follower feeds rank the same items with
    // scores equal to 1e-6 (the follower replayed the exact same WAL events).
    let leader_feed: serde_json::Value = client
        .get(format!("{leader_base}/feed?profile=trending&limit=8"))
        .send()
        .unwrap()
        .json()
        .unwrap();
    let follower_feed: serde_json::Value = client
        .get(format!("{follower_base}/feed?profile=trending&limit=8"))
        .send()
        .unwrap()
        .json()
        .unwrap();
    let l_items = leader_feed["items"].as_array().unwrap();
    let f_items = follower_feed["items"].as_array().unwrap();
    assert!(!f_items.is_empty(), "follower must serve replicated items");
    assert_eq!(l_items.len(), f_items.len(), "same number of ranked items");

    let mut l_scores: std::collections::HashMap<u64, f64> = std::collections::HashMap::new();
    for it in l_items {
        l_scores.insert(
            it["entity_id"].as_u64().unwrap(),
            it["score"].as_f64().unwrap(),
        );
    }
    for it in f_items {
        let id = it["entity_id"].as_u64().unwrap();
        let f = it["score"].as_f64().unwrap();
        let l = *l_scores.get(&id).expect("follower item also on leader");
        assert!((l - f).abs() < 1e-6, "entity {id}: leader={l} follower={f}");
    }

    rt.shutdown_timeout(Duration::from_secs(2));
}

/// m12p1: the region node serves `POST /vector_search` over a corpus whose
/// embeddings replicated to the follower over real gRPC. Embeddings written to
/// the leader (forwarded + WAL-replicated, kind-2) are INDEXED on the follower,
/// and a k-NN query on the FOLLOWER returns the nearest item closest-first.
#[test]
fn region_node_serves_vector_search_over_replicated_corpus() {
    let pair = Pair::new();

    let leader_dir = region_dir();
    let leader = build_region_emb(pair.topology(), &pair.leader_name, &leader_dir);
    let follower_dir = region_dir();
    let follower = build_region_emb(pair.topology(), &pair.follower_name, &follower_dir);

    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .unwrap();
    serve(
        &rt,
        build_region_router(Arc::new(leader), mk_test_creds()),
        pair.leader_http,
    );
    serve(
        &rt,
        build_region_router(Arc::new(follower), mk_test_creds()),
        pair.follower_http,
    );

    let client = reqwest::blocking::Client::new();
    let leader_base = format!("http://{}", pair.leader_http);
    let follower_base = format!("http://{}", pair.follower_http);

    // 8 items on distinct axes (item i → unit vector on axis i-1). Items are not
    // WAL-replicated in this mode, so broadcast them to both nodes; embeddings ARE
    // WAL-replicated, so write them only on the leader and let them flow to the
    // follower.
    let axis_vec = |axis: usize| -> Vec<f32> {
        let mut v = vec![0.0_f32; 8];
        v[axis] = 1.0;
        v
    };
    for i in 1..=8u64 {
        for base in [&leader_base, &follower_base] {
            let resp = client
                .post(format!("{base}/items"))
                .json(&serde_json::json!({
                    "entity_id": i,
                    "metadata": { "title": format!("item {i}") }
                }))
                .send()
                .unwrap();
            assert!(resp.status().is_success(), "POST /items: {}", resp.status());
        }
        let resp = client
            .post(format!("{leader_base}/embeddings"))
            .json(&serde_json::json!({ "entity_id": i, "values": axis_vec((i - 1) as usize) }))
            .send()
            .unwrap();
        assert!(
            resp.status().is_success(),
            "POST /embeddings on leader: {}",
            resp.status()
        );
    }

    // Follower converges: all 8 embedding events applied, lag back to 0.
    poll_status(&client, &follower_base, |applied, lag| {
        applied >= 8 && lag == 0
    });

    // k-NN on the FOLLOWER: an axis-0 query must return item 1 first (its exact
    // vector), with k honored and distances ascending.
    let resp = client
        .post(format!("{follower_base}/vector_search"))
        .json(&serde_json::json!({ "vector": axis_vec(0), "k": 3 }))
        .send()
        .unwrap();
    assert_eq!(
        resp.status(),
        reqwest::StatusCode::OK,
        "POST /vector_search on follower"
    );
    let body: serde_json::Value = resp.json().unwrap();
    let items = body["items"].as_array().expect("items array");
    assert_eq!(items.len(), 3, "k=3 nearest from the follower");
    assert_eq!(
        items[0]["entity_id"].as_u64(),
        Some(1),
        "the axis-0 item is nearest on the replicated follower; got {body}"
    );
    let dists: Vec<f64> = items
        .iter()
        .map(|it| it["distance"].as_f64().unwrap())
        .collect();
    for w in dists.windows(2) {
        assert!(w[0] <= w[1], "distances must ascend: {dists:?}");
    }

    rt.shutdown_timeout(Duration::from_secs(2));
}

/// m11p6 sharding × replication (in-process): 2 nodes × 2 shards × RF=2, with
/// shard 0 led by node A and shard 1 led by node B (balanced leaders). Every
/// node hosts a replica of BOTH groups. A write routes by entity hash to its
/// shard's leader (A applies its shard-0 writes locally and FORWARDS its
/// shard-1 writes to B, and vice versa — the unified path, no /sharded needed);
/// each shard replicates to its follower on the other node; and a /feed read on
/// either node scatters over both local groups and merges, returning items from
/// BOTH shards. This is the exit-gate shape at 2×2 — proof the
/// replicated-XOR-sharded split is gone.
#[test]
fn region_sharded_writes_route_per_shard_and_reads_scatter() {
    // 2 nodes, 4 gRPC ports (one per (node, shard)), 2 HTTP ports.
    let names = ["node-a".to_string(), "node-b".to_string()];
    let http = [free_addr(), free_addr()];
    // grpc[node][shard]
    let grpc = [[free_addr(), free_addr()], [free_addr(), free_addr()]];

    let topology = || -> TopologySpec {
        let regions = (0..2)
            .map(|i| RegionSpec {
                name: names[i].clone(),
                // The node's base grpc_addr is shard 0's; shard 1 sets its own.
                grpc_addr: Some(grpc[i][0].to_string()),
                grpc_bind: None,
                http_addr: Some(http[i].to_string()),
                grpc_tls: None,
                metrics_addr: None,
                zone: None,
            })
            .collect();
        let shard = |id: u16, leader: usize| ShardSpec {
            id,
            leader: Some(names[leader].clone()),
            replicas: (0..2)
                .map(|n| ShardReplicaSpec {
                    node: names[n].clone(),
                    grpc_addr: Some(grpc[n][id as usize].to_string()),
                    grpc_bind: None,
                })
                .collect(),
        };
        TopologySpec {
            regions,
            leader: names[0].clone(), // legacy field unused once shards: is set
            write_workers: None,
            timeouts: TimeoutsSpec::default(),
            replication: ReplicationSpec::default(),
            wal: WalSpec::default(),
            election: ElectionSpec::default(),
            shards: Some(vec![shard(0, 0), shard(1, 1)]),
        }
    };

    let dirs: Vec<tempfile::TempDir> = (0..2).map(|_| region_dir()).collect();
    let node_a = Arc::new(build_region(topology(), &names[0], &dirs[0]));
    let node_b = Arc::new(build_region(topology(), &names[1], &dirs[1]));

    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(4)
        .enable_all()
        .build()
        .unwrap();
    serve(
        &rt,
        build_region_router(Arc::clone(&node_a), mk_test_creds()),
        http[0],
    );
    serve(
        &rt,
        build_region_router(Arc::clone(&node_b), mk_test_creds()),
        http[1],
    );

    let client = reqwest::blocking::Client::new();
    let base_a = format!("http://{}", http[0]);
    let base_b = format!("http://{}", http[1]);

    // Partition entity ids by their shard so we can assert BOTH shards receive
    // writes (the same FNV-1a router the gateway uses).
    let router = ShardRouter::hash(2).unwrap();
    let mut shard0_ids = Vec::new();
    let mut shard1_ids = Vec::new();
    for i in 1..=24u64 {
        match router.route(EntityId::new(i)).0 {
            0 => shard0_ids.push(i),
            _ => shard1_ids.push(i),
        }
    }
    assert!(
        !shard0_ids.is_empty() && !shard1_ids.is_empty(),
        "the hash must spread 24 ids across both shards (got {} / {})",
        shard0_ids.len(),
        shard1_ids.len()
    );

    // Write every item + a view signal through node A ONLY. A leads shard 0
    // (applies locally) and follows shard 1 (forwards to B) — one client, one
    // endpoint, both shards.
    for i in 1..=24u64 {
        let resp = client
            .post(format!("{base_a}/items"))
            .json(&serde_json::json!({
                "entity_id": i, "metadata": { "title": format!("item {i}") }
            }))
            .send()
            .unwrap();
        assert!(
            resp.status().is_success(),
            "POST /items id={i}: {}",
            resp.status()
        );
        let resp = client
            .post(format!("{base_a}/signals"))
            .json(&serde_json::json!({ "entity_id": i, "signal": "view", "weight": 1.0 }))
            .send()
            .unwrap();
        assert!(
            resp.status().is_success(),
            "POST /signals id={i}: {}",
            resp.status()
        );
    }

    // Both nodes' feeds must, after convergence, return EVERY item — proof that
    // (1) A forwarded shard-1 writes to B, (2) each shard replicated to its
    // follower, and (3) the read scatters over both local groups and merges.
    let feed_ids = |base: &str| -> std::collections::HashSet<u64> {
        let feed: serde_json::Value = client
            .get(format!("{base}/feed?profile=trending&limit=50"))
            .send()
            .unwrap()
            .json()
            .unwrap();
        feed["items"]
            .as_array()
            .unwrap()
            .iter()
            .filter_map(|it| it["entity_id"].as_u64())
            .collect()
    };

    let deadline = Instant::now() + Duration::from_secs(10);
    loop {
        let a = feed_ids(&base_a);
        let b = feed_ids(&base_b);
        let all: std::collections::HashSet<u64> = (1..=24u64).collect();
        if a == all && b == all {
            // Sanity: both shards are actually represented (not one big shard).
            assert!(
                shard0_ids.iter().all(|i| a.contains(i))
                    && shard1_ids.iter().all(|i| a.contains(i)),
                "node A feed must merge BOTH shards"
            );
            break;
        }
        assert!(
            Instant::now() <= deadline,
            "feeds did not converge to all 24 items across both shards within 10s \
             (A={}, B={})",
            a.len(),
            b.len()
        );
        std::thread::sleep(Duration::from_millis(50));
    }

    rt.shutdown_timeout(Duration::from_secs(2));
}

/// m11p6 PARTIAL placement (the `EntityRoute::Remote` path + group-local reads):
/// 3 nodes, 2 shards, RF=2, placed so shard 0 = {node-a (leader), node-b} and
/// shard 1 = {node-b, node-c (leader)}. So node-a hosts ONLY shard 0, node-c
/// ONLY shard 1, node-b BOTH. Writing every entity through node-a forces its
/// shard-1 writes down the not-hosted (`Remote`) path — forwarded leader-first to
/// a shard-1 replica node, which applies. node-b (full placement) then merges the
/// whole corpus; node-a / node-c (partial) feed only their own group — the
/// documented L2 read scope, asserted so it is intentional, not a silent gap.
#[test]
fn region_sharded_subset_placement_forwards_and_reads_are_group_scoped() {
    let names = [
        "node-a".to_string(),
        "node-b".to_string(),
        "node-c".to_string(),
    ];
    let http = [free_addr(), free_addr(), free_addr()];
    let (a_s0, b_s0, b_s1, c_s1) = (free_addr(), free_addr(), free_addr(), free_addr());

    let topology = || -> TopologySpec {
        // Each node's RegionSpec base grpc_addr (a hosted addr); replicas are
        // explicit per (node, shard), so the base is only the declared advertise.
        let bases = [a_s0, b_s0, c_s1];
        let regions = (0..3)
            .map(|i| RegionSpec {
                name: names[i].clone(),
                grpc_addr: Some(bases[i].to_string()),
                grpc_bind: None,
                http_addr: Some(http[i].to_string()),
                grpc_tls: None,
                metrics_addr: None,
                zone: None,
            })
            .collect();
        let replica = |node: usize, addr: SocketAddr| ShardReplicaSpec {
            node: names[node].clone(),
            grpc_addr: Some(addr.to_string()),
            grpc_bind: None,
        };
        TopologySpec {
            regions,
            leader: names[0].clone(),
            write_workers: None,
            timeouts: TimeoutsSpec::default(),
            replication: ReplicationSpec::default(),
            wal: WalSpec::default(),
            election: ElectionSpec::default(),
            shards: Some(vec![
                ShardSpec {
                    id: 0,
                    leader: Some(names[0].clone()),
                    replicas: vec![replica(0, a_s0), replica(1, b_s0)],
                },
                ShardSpec {
                    id: 1,
                    leader: Some(names[1].clone()),
                    replicas: vec![replica(1, b_s1), replica(2, c_s1)],
                },
            ]),
        }
    };

    let dirs: Vec<tempfile::TempDir> = (0..3).map(|_| region_dir()).collect();
    let node_a = Arc::new(build_region(topology(), &names[0], &dirs[0]));
    let node_b = Arc::new(build_region(topology(), &names[1], &dirs[1]));
    let node_c = Arc::new(build_region(topology(), &names[2], &dirs[2]));

    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(4)
        .enable_all()
        .build()
        .unwrap();
    serve(
        &rt,
        build_region_router(Arc::clone(&node_a), mk_test_creds()),
        http[0],
    );
    serve(
        &rt,
        build_region_router(Arc::clone(&node_b), mk_test_creds()),
        http[1],
    );
    serve(
        &rt,
        build_region_router(Arc::clone(&node_c), mk_test_creds()),
        http[2],
    );

    let client = reqwest::blocking::Client::new();
    let base_a = format!("http://{}", http[0]);
    let base_b = format!("http://{}", http[1]);
    let base_c = format!("http://{}", http[2]);

    let router = ShardRouter::hash(2).unwrap();
    let mut shard0_ids = std::collections::HashSet::new();
    let mut shard1_ids = std::collections::HashSet::new();
    for i in 1..=24u64 {
        if router.route(EntityId::new(i)).0 == 0 {
            shard0_ids.insert(i);
        } else {
            shard1_ids.insert(i);
        }
    }
    assert!(
        !shard0_ids.is_empty() && !shard1_ids.is_empty(),
        "the hash must spread 24 ids across both shards"
    );

    // Write every item + a view signal through node-A ONLY. Shard-0 ids apply
    // locally (A leads shard 0); shard-1 ids take EntityRoute::Remote (A hosts no
    // shard-1 replica) and forward to a shard-1 replica node — proof the
    // not-hosted gateway path works for both items and signals.
    for i in 1..=24u64 {
        let item = client
            .post(format!("{base_a}/items"))
            .json(&serde_json::json!({ "entity_id": i, "metadata": { "title": format!("item {i}") } }))
            .send()
            .unwrap();
        assert!(
            item.status().is_success(),
            "POST /items id={i} via node-a: {}",
            item.status()
        );
        let sig = client
            .post(format!("{base_a}/signals"))
            .json(&serde_json::json!({ "entity_id": i, "signal": "view", "weight": 1.0 }))
            .send()
            .unwrap();
        assert!(
            sig.status().is_success(),
            "POST /signals id={i} via node-a: {}",
            sig.status()
        );
    }

    let feed_ids = |base: &str| -> std::collections::HashSet<u64> {
        let feed: serde_json::Value = client
            .get(format!("{base}/feed?profile=trending&limit=50"))
            .send()
            .unwrap()
            .json()
            .unwrap();
        feed["items"]
            .as_array()
            .unwrap()
            .iter()
            .filter_map(|it| it["entity_id"].as_u64())
            .collect()
    };
    let all: std::collections::HashSet<u64> = (1..=24u64).collect();

    let deadline = Instant::now() + Duration::from_secs(10);
    loop {
        let a = feed_ids(&base_a);
        let b = feed_ids(&base_b);
        let c = feed_ids(&base_c);
        // m12p4 cross-shard unified reads: EVERY node now returns the WHOLE corpus.
        // node-b hosts both groups (local scatter covers it); node-a hosts only
        // group 0 and node-c only group 1, but each cross-shard fans out to the
        // group it does not host, so all three converge to the full 24-id set.
        // (Pre-m12p4, node-a returned only `shard0_ids` and node-c only
        // `shard1_ids` — the local-shard-only limitation the L4 fan-out closes.)
        if a == all && b == all && c == all {
            break;
        }
        assert!(
            Instant::now() <= deadline,
            "subset-placement cross-shard feeds did not converge to the whole corpus \
             (a={}/24, b={}/24, c={}/24)",
            a.len(),
            b.len(),
            c.len()
        );
        std::thread::sleep(Duration::from_millis(50));
    }

    rt.shutdown_timeout(Duration::from_secs(2));
}

/// A write to the FOLLOWER (a non-leader node) is FORWARDED to the leader; with
/// the leader process not running, the forward fails and degrades to a 503 whose
/// body names the (unreachable) leader — the task-03 leader-unreachable contract.
#[test]
fn region_node_rejects_writes_when_not_leader() {
    let pair = Pair::new();

    let follower_dir = region_dir();
    let follower = build_region(pair.topology(), &pair.follower_name, &follower_dir);
    // Build the runtime AFTER the nodes (and their data-dir guards):
    // locals drop in reverse order, so the runtime — which owns the
    // nodes via the serve tasks — tears down BEFORE the dirs delete,
    // on the panic-unwind path too (m11p2: nodes are persistent now).
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .unwrap();
    serve(
        &rt,
        build_region_router(Arc::new(follower), mk_test_creds()),
        pair.follower_http,
    );

    let client = reqwest::blocking::Client::new();
    let follower_base = format!("http://{}", pair.follower_http);

    let resp = client
        .post(format!("{follower_base}/signals"))
        .json(&serde_json::json!({ "entity_id": 1, "signal": "view", "weight": 1.0 }))
        .send()
        .unwrap();
    assert_eq!(
        resp.status().as_u16(),
        503,
        "a non-leader write must be 503 NotLeader"
    );
    let body: serde_json::Value = resp.json().unwrap();
    assert_eq!(
        body["leader"].as_str(),
        Some(pair.leader_name.as_str()),
        "the 503 body must name the leader: {body}"
    );

    rt.shutdown_timeout(Duration::from_secs(2));
}

/// Partition the follower → leader ships are skipped (follower lags) → heal →
/// the leader redelivers over gRPC and the follower converges. A SECOND heal is
/// a no-op (idempotent): the follower's scores are unchanged.
#[test]
fn region_node_partition_heal() {
    let pair = Pair::new();

    let leader_dir = region_dir();
    let leader = build_region(pair.topology(), &pair.leader_name, &leader_dir);
    let follower_dir = region_dir();
    let follower = build_region(pair.topology(), &pair.follower_name, &follower_dir);
    // Build the runtime AFTER the nodes (and their data-dir guards):
    // locals drop in reverse order, so the runtime — which owns the
    // nodes via the serve tasks — tears down BEFORE the dirs delete,
    // on the panic-unwind path too (m11p2: nodes are persistent now).
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .unwrap();
    serve(
        &rt,
        build_region_router(Arc::new(leader), mk_test_creds()),
        pair.leader_http,
    );
    serve(
        &rt,
        build_region_router(Arc::new(follower), mk_test_creds()),
        pair.follower_http,
    );

    let client = reqwest::blocking::Client::new();
    let leader_base = format!("http://{}", pair.leader_http);
    let follower_base = format!("http://{}", pair.follower_http);

    let post_signal = |entity: u64| {
        let resp = client
            .post(format!("{leader_base}/signals"))
            .json(&serde_json::json!({ "entity_id": entity, "signal": "view", "weight": 1.0 }))
            .send()
            .unwrap();
        assert!(
            resp.status().is_success(),
            "leader signal: {}",
            resp.status()
        );
    };
    let post_item = |entity: u64| {
        let resp = client
            .post(format!("{leader_base}/items"))
            .json(&serde_json::json!({ "entity_id": entity, "metadata": { "t": entity.to_string() } }))
            .send()
            .unwrap();
        assert!(resp.status().is_success(), "leader item: {}", resp.status());
    };

    // Create items 1..=5 on the LEADER up front, so heal can backfill the ones the
    // follower misses while partitioned (BUG 3). Write 2 signals, let the follower
    // catch up.
    for e in 1..=5u64 {
        post_item(e);
    }
    post_signal(1);
    post_signal(2);
    // m11p2: items ride the WAL too — 5 item records + 2 signals = 7 seqnos.
    poll_status(&client, &follower_base, |applied, _| applied >= 7);

    // Partition the follower from the leader, then write 3 more.
    let resp = client
        .post(format!("{leader_base}/cluster/partition"))
        .json(&serde_json::json!({ "region": pair.follower_name }))
        .send()
        .unwrap();
    assert!(resp.status().is_success(), "partition: {}", resp.status());
    for e in 3..=5u64 {
        post_signal(e);
    }
    // While partitioned the follower is STUCK at applied=7: the leader's
    // ships were skipped, so no segments arrive. (Its locally-reported lag stays
    // 0 because a follower cannot observe the leader's progress while
    // partitioned — cross-node lag aggregation is task 03; the load-bearing
    // proof here is that `applied` does NOT advance past 2.)
    let lagging = poll_status(&client, &follower_base, |applied, _| applied == 7);
    assert_eq!(lagging["applied_events"].as_u64(), Some(7));
    // Give the leader a beat to (not) ship — applied must remain 7, proving the
    // partition truly skips the eager ships rather than racing convergence.
    std::thread::sleep(Duration::from_millis(200));
    let still: serde_json::Value = client
        .get(format!("{follower_base}/cluster/status/local"))
        .send()
        .unwrap()
        .json()
        .unwrap();
    assert_eq!(
        still["applied_events"].as_u64(),
        Some(7),
        "partitioned follower must NOT receive the leader's post-partition writes"
    );

    // Heal: the leader redelivers the missed segments over gRPC.
    let heal = |base: &str| {
        let resp = client
            .post(format!("{base}/cluster/heal"))
            .json(&serde_json::json!({ "region": pair.follower_name }))
            .send()
            .unwrap();
        assert!(resp.status().is_success(), "heal: {}", resp.status());
    };
    heal(&leader_base);
    poll_status(&client, &follower_base, |applied, lag| {
        applied >= 10 && lag == 0
    });

    // Capture the follower's converged decay for entity 5. Heal redelivered the
    // signal AND backfilled the item (BUG 3), so the follower can rank entity 5
    // WITHOUT the test posting anything. Poll until the score stabilizes (two equal
    // consecutive non-zero reads) so the capture does not race the async item
    // index; after it settles the score is stable, so re-heal idempotence is exact.
    let score_of = |entity: u64| -> f64 {
        let deadline = Instant::now() + Duration::from_secs(10);
        let mut last = -1.0f64;
        loop {
            let s = feed_score(&client, &follower_base, entity);
            if s > 0.0 && (s - last).abs() < 1e-12 {
                return s;
            }
            last = s;
            assert!(
                Instant::now() <= deadline,
                "heal must deliver signal+item for entity {entity} so the follower ranks it"
            );
            std::thread::sleep(Duration::from_millis(50));
        }
    };
    let before = score_of(5);

    // Second heal is idempotent: re-ships nothing new (follower already applied
    // through seq 10) and re-broadcasts the same items, so the score is unchanged.
    heal(&leader_base);
    poll_status(&client, &follower_base, |applied, lag| {
        applied == 10 && lag == 0
    });
    let after = score_of(5);
    assert!(
        (before - after).abs() < 1e-9,
        "idempotent re-heal must not change scores: before={before} after={after}"
    );

    rt.shutdown_timeout(Duration::from_secs(2));
}

/// Promote flips the leadership view: the OLD leader now rejects writes
/// (NotLeader), the NEW leader accepts and ships, and the always-on receiver on
/// the demoted node applies the new leader's segments.
#[test]
fn region_node_promote_local() {
    let pair = Pair::new();

    let leader_dir = region_dir();
    let leader = build_region(pair.topology(), &pair.leader_name, &leader_dir);
    let follower_dir = region_dir();
    let follower = build_region(pair.topology(), &pair.follower_name, &follower_dir);
    // Build the runtime AFTER the nodes (and their data-dir guards):
    // locals drop in reverse order, so the runtime — which owns the
    // nodes via the serve tasks — tears down BEFORE the dirs delete,
    // on the panic-unwind path too (m11p2: nodes are persistent now).
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .unwrap();
    serve(
        &rt,
        build_region_router(Arc::new(leader), mk_test_creds()),
        pair.leader_http,
    );
    serve(
        &rt,
        build_region_router(Arc::new(follower), mk_test_creds()),
        pair.follower_http,
    );

    let client = reqwest::blocking::Client::new();
    let leader_base = format!("http://{}", pair.leader_http);
    let follower_base = format!("http://{}", pair.follower_http);

    // Promote the FOLLOWER to leader on BOTH nodes (each node holds its own view).
    for base in [&leader_base, &follower_base] {
        let resp = client
            .post(format!("{base}/cluster/promote"))
            .json(&serde_json::json!({ "region": pair.follower_name }))
            .send()
            .unwrap();
        assert!(resp.status().is_success(), "promote: {}", resp.status());
    }

    // The OLD leader (us-east), now demoted, transparently FORWARDS a write to the
    // new leader (eu-west) and relays its 204 (task 03 replaces the standalone
    // NotLeader 503 with leader forwarding). Item 1 must exist on eu-west to be
    // rankable, but the forwarded signal itself proves the forward path.
    let _ = client
        .post(format!("{follower_base}/items"))
        .json(&serde_json::json!({ "entity_id": 1, "metadata": {} }))
        .send()
        .unwrap();
    let resp = client
        .post(format!("{leader_base}/signals"))
        .json(&serde_json::json!({ "entity_id": 1, "signal": "view", "weight": 1.0 }))
        .send()
        .unwrap();
    assert_eq!(
        resp.status().as_u16(),
        204,
        "demoted node must FORWARD the write to the new leader (204), not 503: {}",
        resp.status()
    );

    // The NEW leader (eu-west) accepts and ships to the demoted node (us-east),
    // whose ALWAYS-ON receiver applies the segments. Broadcast items to both.
    for e in 1..=4u64 {
        for base in [&leader_base, &follower_base] {
            let _ = client
                .post(format!("{base}/items"))
                .json(&serde_json::json!({ "entity_id": e, "metadata": {} }))
                .send()
                .unwrap();
        }
        let resp = client
            .post(format!("{follower_base}/signals"))
            .json(&serde_json::json!({ "entity_id": e, "signal": "view", "weight": 1.0 }))
            .send()
            .unwrap();
        assert!(
            resp.status().is_success(),
            "new-leader signal: {}",
            resp.status()
        );
    }

    // The demoted node (us-east) applies the new leader's stream — proving the
    // always-on receiver runs on every node, not just the initial follower.
    poll_status(&client, &leader_base, |applied, lag| {
        applied >= 4 && lag == 0
    });

    rt.shutdown_timeout(Duration::from_secs(2));
}

/// `POST /hardnegs` records a hide on the local node (node-local by design).
#[test]
fn region_node_records_hardneg() {
    let pair = Pair::new();

    let leader_dir = region_dir();
    let leader = build_region(pair.topology(), &pair.leader_name, &leader_dir);
    // Build the runtime AFTER the nodes (and their data-dir guards):
    // locals drop in reverse order, so the runtime — which owns the
    // nodes via the serve tasks — tears down BEFORE the dirs delete,
    // on the panic-unwind path too (m11p2: nodes are persistent now).
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .unwrap();
    serve(
        &rt,
        build_region_router(Arc::new(leader), mk_test_creds()),
        pair.leader_http,
    );

    let client = reqwest::blocking::Client::new();
    let leader_base = format!("http://{}", pair.leader_http);

    let resp = client
        .post(format!("{leader_base}/hardnegs"))
        .json(&serde_json::json!({ "user_id": 42, "item_id": 7 }))
        .send()
        .unwrap();
    assert_eq!(
        resp.status().as_u16(),
        204,
        "POST /hardnegs must record a hide: {}",
        resp.status()
    );

    rt.shutdown_timeout(Duration::from_secs(2));
}

// ── BUG 1: lag gauge across a leadership change ──────────────────────────────

/// A fully-declared THREE-region topology pointing at three loopback addresses.
struct Trio {
    names: [String; 3],
    grpc: [SocketAddr; 3],
    http: [SocketAddr; 3],
}

impl Trio {
    fn new() -> Self {
        Self {
            names: ["us-east".into(), "eu-west".into(), "ap-south".into()],
            grpc: [free_addr(), free_addr(), free_addr()],
            http: [free_addr(), free_addr(), free_addr()],
        }
    }

    fn topology(&self) -> TopologySpec {
        TopologySpec {
            regions: (0..3)
                .map(|i| RegionSpec {
                    name: self.names[i].clone(),
                    grpc_addr: Some(self.grpc[i].to_string()),
                    grpc_bind: None,
                    http_addr: Some(self.http[i].to_string()),
                    grpc_tls: None,
                    metrics_addr: None,
                    zone: None,
                })
                .collect(),
            leader: self.names[0].clone(),
            write_workers: None,
            timeouts: TimeoutsSpec::default(),
            replication: ReplicationSpec::default(),
            wal: WalSpec::default(),
            election: ElectionSpec::default(),
            shards: None,
        }
    }
}

/// BUG 3 reproduction: items are HTTP-broadcast (not WAL-relayed), so a follower
/// that is partitioned during the item broadcast misses the items forever — and
/// heal (which only re-ships signal segments) must ALSO backfill the item
/// metadata + embeddings, so heal is the single recovery verb that leaves the
/// follower with EXACTLY the leader's data.
///
/// Before the fix, the leader's `/items` broadcast to a partitioned follower
/// landed in the `failed` list and nothing backfilled it; the follower's feed
/// could not rank items it never received, even after heal closed the signal gap.
#[test]
fn region_node_heal_backfills_missed_items() {
    let pair = Pair::new();

    let leader_dir = region_dir();
    let leader = build_region(pair.topology(), &pair.leader_name, &leader_dir);
    let follower_node_dir = region_dir();
    let follower = Arc::new(build_region(
        pair.topology(),
        &pair.follower_name,
        &follower_node_dir,
    ));
    // Build the runtime AFTER the nodes (and their data-dir guards):
    // locals drop in reverse order, so the runtime — which owns the
    // nodes via the serve tasks — tears down BEFORE the dirs delete,
    // on the panic-unwind path too (m11p2: nodes are persistent now).
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .unwrap();
    serve(
        &rt,
        build_region_router(Arc::new(leader), mk_test_creds()),
        pair.leader_http,
    );
    // The follower's HTTP server is NOT serving yet — it models a region that is
    // DOWN during the leader's item/signal broadcast. The leader's best-effort
    // HTTP broadcast to it will fail (connection refused) and land in `failed`,
    // exactly as it would for a crashed/restarting region. The follower's gRPC
    // receiver IS running (started in ShardReplica::new), so once the
    // leader heals it, the relay re-ships signals — but the item broadcast that
    // failed during downtime is what heal must backfill.

    let client = reqwest::blocking::Client::new();
    let leader_base = format!("http://{}", pair.leader_http);
    let follower_base = format!("http://{}", pair.follower_http);

    // Partition the follower at the RELAY level too, so the eager signal ships
    // are skipped while it is down (mirrors the real partition: no gRPC either).
    let resp = client
        .post(format!("{leader_base}/cluster/partition"))
        .json(&serde_json::json!({ "region": pair.follower_name }))
        .send()
        .unwrap();
    assert!(resp.status().is_success(), "partition: {}", resp.status());

    // Leader writes items + signals for entities 6,7,8 while the follower is DOWN.
    // The item broadcast to the follower fails (connection refused); the signal
    // ship is skipped (partitioned).
    for e in 6..=8u64 {
        let resp = client
            .post(format!("{leader_base}/items"))
            .json(&serde_json::json!({
                "entity_id": e, "metadata": { "title": format!("item {e}") }
            }))
            .send()
            .unwrap();
        assert!(resp.status().is_success(), "leader item: {}", resp.status());
        let resp = client
            .post(format!("{leader_base}/signals"))
            .json(&serde_json::json!({ "entity_id": e, "signal": "view", "weight": 1.0 }))
            .send()
            .unwrap();
        assert!(
            resp.status().is_success(),
            "leader signal: {}",
            resp.status()
        );
    }

    // The follower comes back up (HTTP server starts serving). Its store is empty
    // for these items — the broadcast during its downtime was lost.
    serve(
        &rt,
        build_region_router(Arc::clone(&follower), mk_test_creds()),
        pair.follower_http,
    );

    // The recovered follower knows none of these items: its feed is empty.
    let pre_feed: serde_json::Value = client
        .get(format!("{follower_base}/feed?profile=trending&limit=10"))
        .send()
        .unwrap()
        .json()
        .unwrap();
    assert!(
        pre_feed["items"].as_array().unwrap().is_empty(),
        "recovered follower must not know the leader's items yet (broadcast lost during \
         downtime): {pre_feed}"
    );

    // Heal: the leader re-ships the missed signal segments AND must backfill the
    // missed item metadata. After heal, the follower has EXACTLY the leader's
    // data — its feed ranks items 6,7,8 WITHOUT the test re-posting them.
    let resp = client
        .post(format!("{leader_base}/cluster/heal"))
        .json(&serde_json::json!({ "region": pair.follower_name }))
        .send()
        .unwrap();
    assert!(resp.status().is_success(), "heal: {}", resp.status());

    // Converge on the signal HWM (3 segments), then assert item parity.
    poll_status(&client, &follower_base, |applied, lag| {
        applied >= 3 && lag == 0
    });

    // Poll the follower's feed until the backfilled items appear (heal's item
    // re-broadcast is async over HTTP).
    let deadline = Instant::now() + Duration::from_secs(5);
    loop {
        let feed: serde_json::Value = client
            .get(format!("{follower_base}/feed?profile=trending&limit=10"))
            .send()
            .unwrap()
            .json()
            .unwrap();
        let ids: std::collections::HashSet<u64> = feed["items"]
            .as_array()
            .unwrap()
            .iter()
            .filter_map(|it| it["entity_id"].as_u64())
            .collect();
        if [6u64, 7, 8].iter().all(|e| ids.contains(e)) {
            break;
        }
        assert!(
            Instant::now() <= deadline,
            "heal must backfill items 6,7,8 to the follower's feed; saw {ids:?}"
        );
        std::thread::sleep(Duration::from_millis(20));
    }

    rt.shutdown_timeout(Duration::from_secs(2));
}

/// BUG 1 reproduction: after the leader stream populates a follower's lag gauge
/// (HWM = N for the OLD leader's shard), promoting a DIFFERENT region to leader
/// must NOT leave the now-non-leader follower reporting a permanent stale lag.
///
/// Before the fix, `local_status` computed
/// `lag = lag_gauge.leader_seqno() − applied_seqno(NEW-leader-shard)`. The gauge's
/// `leader_seqno` is a single monotonic scalar fed by EVERY source stream, so it
/// still held the OLD leader's HWM (N), while `applied_seqno(new-leader-shard)`
/// was 0 (the new leader had not shipped). A fully-converged follower then
/// reported `lag = N − 0 = N` forever.
#[test]
fn region_node_lag_honest_across_promote() {
    // Number of signals the old leader ships before the leadership change.
    const N: u64 = 6;

    let trio = Trio::new();

    // Three nodes: us-east (leader, shard 0), eu-west (shard 1), ap-south
    // (shard 2). Dirs FIRST (see `region_dir` for the drop-order contract).
    let dirs: Vec<tempfile::TempDir> = (0..3).map(|_| region_dir()).collect();
    let nodes: Vec<Arc<ClusterNode>> = (0..3)
        .map(|i| Arc::new(build_region(trio.topology(), &trio.names[i], &dirs[i])))
        .collect();
    // Build the runtime AFTER the nodes (and their data-dir guards): locals
    // drop in reverse order, so the runtime — which owns the nodes via the
    // serve tasks — tears down BEFORE the dirs delete, on the panic-unwind
    // path too (m11p2: nodes are persistent now).
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(3)
        .enable_all()
        .build()
        .unwrap();
    for (i, node) in nodes.iter().enumerate() {
        serve(
            &rt,
            build_region_router(Arc::clone(node), mk_test_creds()),
            trio.http[i],
        );
    }

    let client = reqwest::blocking::Client::new();
    let bases: Vec<String> = (0..3).map(|i| format!("http://{}", trio.http[i])).collect();

    // us-east (shard 0) leads: write N signals, replicated to eu-west + ap-south.
    // This populates each follower's lag gauge HWM for the OLD leader's shard.
    for e in 1..=N {
        for base in &bases {
            let _ = client
                .post(format!("{base}/items"))
                .json(&serde_json::json!({ "entity_id": e, "metadata": {} }))
                .send()
                .unwrap();
        }
        let resp = client
            .post(format!("{}/signals", bases[0]))
            .json(&serde_json::json!({ "entity_id": e, "signal": "view", "weight": 1.0 }))
            .send()
            .unwrap();
        assert!(
            resp.status().is_success(),
            "leader signal: {}",
            resp.status()
        );
    }

    // eu-west (the future non-leader) fully converges on the old leader's stream.
    poll_status(&client, &bases[1], |applied, lag| applied >= N && lag == 0);

    // Promote ap-south (shard 2) to leader on ALL nodes. ap-south has shipped
    // NOTHING yet, so every node's applied-seqno for shard 2 is 0.
    for base in &bases {
        let resp = client
            .post(format!("{base}/cluster/promote"))
            .json(&serde_json::json!({ "region": trio.names[2] }))
            .send()
            .unwrap();
        assert!(resp.status().is_success(), "promote: {}", resp.status());
    }

    // eu-west is now a NON-leader, fully converged (nothing new to apply). Its lag
    // MUST be 0 — it is not behind the new leader, which has shipped nothing.
    // Before the fix this read N (stale gauge HWM from shard 0 − applied(shard2)=0).
    let status: serde_json::Value = client
        .get(format!("{}/cluster/status/local", bases[1]))
        .send()
        .unwrap()
        .json()
        .unwrap();
    assert_eq!(
        status["lag_events"].as_u64(),
        Some(0),
        "a converged non-leader must report lag 0 against a new leader that shipped \
         nothing — stale-gauge lag is bug 1: {status}"
    );

    rt.shutdown_timeout(Duration::from_secs(2));
}

/// m11p3 `ack=quorum` end to end over real gRPC, 2-node shape (majority = the
/// leader plus THE follower):
///
/// 1. With the follower live, a quorum write 204s and carries `x-tidal-seq`.
/// 2. With the follower partitioned (ship-skipped), a quorum write returns
///    the retryable 503 naming the follower as the laggard — while an
///    `x-tidal-ack: leader` override on the SAME cluster still 204s (the
///    leader-ack contract is untouched by a follower outage).
/// 3. After heal, quorum writes 204 again and the leader's `commit_index`
///    catches its `last_seq`.
#[test]
fn region_node_quorum_write_gates_on_follower_durability() {
    let pair = Pair::new();
    // The same spec both nodes parse, with the quorum deployment default and
    // a short budget so the partitioned case fails fast (TopologySpec is not
    // Clone; build it per node).
    let quorum_topology = || {
        let mut t = pair.topology();
        t.replication.ack = Some("quorum".into());
        t.replication.quorum_timeout_ms = Some(400);
        t
    };

    let leader_dir = region_dir();
    let leader = build_region(quorum_topology(), &pair.leader_name, &leader_dir);
    let follower_dir = region_dir();
    let follower = build_region(quorum_topology(), &pair.follower_name, &follower_dir);
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .unwrap();
    serve(
        &rt,
        build_region_router(Arc::new(leader), mk_test_creds()),
        pair.leader_http,
    );
    serve(
        &rt,
        build_region_router(Arc::new(follower), mk_test_creds()),
        pair.follower_http,
    );

    let client = reqwest::blocking::Client::new();
    let leader_base = format!("http://{}", pair.leader_http);

    // ── 1. Live follower: quorum write succeeds with a seq header ──────────
    let resp = client
        .post(format!("{leader_base}/signals"))
        .json(&serde_json::json!({ "entity_id": 1, "signal": "view", "weight": 1.0 }))
        .send()
        .unwrap();
    assert_eq!(
        resp.status().as_u16(),
        204,
        "quorum write with a live follower must succeed"
    );
    let seq: u64 = resp
        .headers()
        .get("x-tidal-seq")
        .expect("a quorum-acked write carries its log seqno")
        .to_str()
        .unwrap()
        .parse()
        .unwrap();
    assert!(seq > 0, "the assigned seqno is a real stream position");

    // The leader's status shows the commit index covering the write.
    let status = poll_status(&client, &leader_base, |_, _| true);
    assert!(
        status["commit_index"].as_u64().unwrap() >= seq,
        "a 204'd quorum write is at or below the commit index: {status}"
    );
    assert_eq!(status["ack"].as_str(), Some("quorum"));

    // ── 2. Partitioned follower: quorum 503 names the laggard ──────────────
    let resp = client
        .post(format!("{leader_base}/cluster/partition"))
        .json(&serde_json::json!({ "region": pair.follower_name }))
        .send()
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200);

    let resp = client
        .post(format!("{leader_base}/signals"))
        .json(&serde_json::json!({ "entity_id": 2, "signal": "view", "weight": 1.0 }))
        .send()
        .unwrap();
    assert_eq!(
        resp.status().as_u16(),
        503,
        "a quorum write cannot commit while THE follower is partitioned (n=2)"
    );
    let body: serde_json::Value = resp.json().unwrap();
    assert_eq!(body["retryable"].as_bool(), Some(true));
    assert_eq!(
        body["laggards"],
        serde_json::json!([pair.follower_name]),
        "the 503 names the laggard: {body}"
    );
    assert_eq!(body["needed"].as_u64(), Some(1));

    // The caller's per-request override still gets leader-ack semantics.
    let resp = client
        .post(format!("{leader_base}/signals"))
        .header("x-tidal-ack", "leader")
        .json(&serde_json::json!({ "entity_id": 3, "signal": "view", "weight": 1.0 }))
        .send()
        .unwrap();
    assert_eq!(
        resp.status().as_u16(),
        204,
        "x-tidal-ack: leader bypasses the quorum gate during the outage"
    );

    // An unknown ack mode is a 400, not a silent default.
    let resp = client
        .post(format!("{leader_base}/signals"))
        .header("x-tidal-ack", "everyone")
        .json(&serde_json::json!({ "entity_id": 4, "signal": "view", "weight": 1.0 }))
        .send()
        .unwrap();
    assert_eq!(resp.status().as_u16(), 400, "invalid x-tidal-ack is a 400");

    // ── 3. Heal: quorum writes commit again ────────────────────────────────
    let resp = client
        .post(format!("{leader_base}/cluster/heal"))
        .json(&serde_json::json!({ "region": pair.follower_name }))
        .send()
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200);

    // Retry the quorum write until the healed pipeline commits one (the heal
    // resume + the durable ack fold may need a retry tick).
    let deadline = Instant::now() + Duration::from_secs(5);
    loop {
        let resp = client
            .post(format!("{leader_base}/signals"))
            .json(&serde_json::json!({ "entity_id": 5, "signal": "view", "weight": 1.0 }))
            .send()
            .unwrap();
        if resp.status().as_u16() == 204 {
            break;
        }
        assert!(
            Instant::now() <= deadline,
            "healed quorum writes must commit within 5s (last: {})",
            resp.status()
        );
        std::thread::sleep(Duration::from_millis(50));
    }
    let status = poll_status(&client, &leader_base, |_, _| true);
    let last_seq = status["last_seq"].as_u64().unwrap();
    let commit = status["commit_index"].as_u64().unwrap();
    assert!(
        commit >= last_seq.saturating_sub(1),
        "post-heal the commit index tracks the flushed frontier: {status}"
    );

    rt.shutdown_timeout(Duration::from_secs(2));
}

/// m11p3: a quorum write THROUGH a follower gateway — the `x-tidal-ack`
/// override travels with the forward, the leader gates on quorum, and the
/// `x-tidal-seq` response header relays back to the original caller. Also
/// covers items + embeddings (kind-1/2 records gate on the same commit index).
#[test]
fn region_node_quorum_forward_and_blob_writes() {
    let pair = Pair::new();

    let leader_dir = region_dir();
    let leader = build_region(pair.topology(), &pair.leader_name, &leader_dir);
    let follower_dir = region_dir();
    let follower = build_region(pair.topology(), &pair.follower_name, &follower_dir);
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .unwrap();
    serve(
        &rt,
        build_region_router(Arc::new(leader), mk_test_creds()),
        pair.leader_http,
    );
    serve(
        &rt,
        build_region_router(Arc::new(follower), mk_test_creds()),
        pair.follower_http,
    );

    let client = reqwest::blocking::Client::new();
    let leader_base = format!("http://{}", pair.leader_http);
    let follower_base = format!("http://{}", pair.follower_http);

    // Quorum item via the FOLLOWER gateway (topology default is leader-ack;
    // the header overrides through the forward).
    let resp = client
        .post(format!("{follower_base}/items"))
        .header("x-tidal-ack", "quorum")
        .json(&serde_json::json!({
            "entity_id": 7,
            "metadata": { "title": "quorum item via follower" }
        }))
        .send()
        .unwrap();
    assert_eq!(
        resp.status().as_u16(),
        201,
        "a forwarded quorum item write must succeed"
    );
    let item_seq: u64 = resp
        .headers()
        .get("x-tidal-seq")
        .expect("the leader's seq header relays through the forward")
        .to_str()
        .unwrap()
        .parse()
        .unwrap();
    assert!(item_seq > 0);

    // Quorum embedding straight at the leader.
    let resp = client
        .post(format!("{leader_base}/embeddings"))
        .header("x-tidal-ack", "quorum")
        .json(&serde_json::json!({ "entity_id": 7, "values": [0.1, 0.2, 0.3, 0.4] }))
        .send()
        .unwrap();
    assert_eq!(resp.status().as_u16(), 204);
    let emb_seq: u64 = resp
        .headers()
        .get("x-tidal-seq")
        .expect("embedding writes carry their seq too")
        .to_str()
        .unwrap()
        .parse()
        .unwrap();
    assert!(
        emb_seq > item_seq,
        "one log: the embedding's seqno follows the item's ({item_seq} -> {emb_seq})"
    );

    // The quorum-acked writes are durable on the follower BY CONTRACT —
    // its applied frontier already covers them (no convergence poll needed,
    // that is the whole point of ack=quorum).
    let follower_status: serde_json::Value = client
        .get(format!("{follower_base}/cluster/status/local"))
        .send()
        .unwrap()
        .json()
        .unwrap();
    assert!(
        follower_status["applied_events"].as_u64().unwrap() >= emb_seq,
        "a 2-node quorum ack means THE follower durably applied it: {follower_status}"
    );

    rt.shutdown_timeout(Duration::from_secs(2));
}

/// m11p5 §3 (tier-2.5): the membership runtime is wired end-to-end through real
/// HTTP and the effective roster. A 2-node cluster with NO conf-change is in
/// era 0: `/cluster/members` returns both regions as VOTERS at version 0, the
/// status surface reports `membership_version: 0` + `membership_role: voter`,
/// and a join dialled at the NON-leader is refused WITH a leader hint (the
/// conf-change gate's `NotLeader` path — the joiner re-targets). This proves the
/// `MembershipView` derives the era-0 roster byte-for-byte from the topology and
/// that the join/members verbs reach the runtime.
#[test]
fn region_membership_era0_roster_and_join_gate() {
    let pair = Pair::new();

    let leader_dir = region_dir();
    let leader = build_region(pair.topology(), &pair.leader_name, &leader_dir);
    let follower_dir = region_dir();
    let follower = build_region(pair.topology(), &pair.follower_name, &follower_dir);
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .unwrap();
    serve(
        &rt,
        build_region_router(Arc::new(leader), mk_test_creds()),
        pair.leader_http,
    );
    serve(
        &rt,
        build_region_router(Arc::new(follower), mk_test_creds()),
        pair.follower_http,
    );

    let client = reqwest::blocking::Client::new();
    let leader_base = format!("http://{}", pair.leader_http);
    let follower_base = format!("http://{}", pair.follower_http);

    // ── Era-0 roster via /cluster/members on the leader ──────────────────────
    let members: serde_json::Value = client
        .get(format!("{leader_base}/cluster/members"))
        .send()
        .unwrap()
        .json()
        .unwrap();
    assert_eq!(
        members["membership_version"].as_u64(),
        Some(0),
        "no conf-change → era 0 (version 0): {members}"
    );
    let roster = members["members"].as_array().expect("members array");
    assert_eq!(roster.len(), 2, "both topology regions are in the roster");
    // Both are voters, with positional ids 0 and 1, names from the topology.
    let by_id: std::collections::HashMap<u64, &serde_json::Value> = roster
        .iter()
        .map(|m| (m["id"].as_u64().unwrap(), m))
        .collect();
    assert_eq!(by_id[&0]["name"].as_str(), Some(pair.leader_name.as_str()));
    assert_eq!(by_id[&0]["role"].as_str(), Some("voter"));
    assert_eq!(
        by_id[&1]["name"].as_str(),
        Some(pair.follower_name.as_str())
    );
    assert_eq!(by_id[&1]["role"].as_str(), Some("voter"));

    // ── Status surface carries the membership fields ─────────────────────────
    let leader_status: serde_json::Value = client
        .get(format!("{leader_base}/cluster/status/local"))
        .send()
        .unwrap()
        .json()
        .unwrap();
    assert_eq!(leader_status["membership_version"].as_u64(), Some(0));
    assert_eq!(
        leader_status["membership_role"].as_str(),
        Some("voter"),
        "an era-0 node is a voter: {leader_status}"
    );

    // The follower's /cluster/members agrees (it derives the SAME era-0 roster
    // from the same topology — the view is internally consistent on every node).
    let follower_members: serde_json::Value = client
        .get(format!("{follower_base}/cluster/members"))
        .send()
        .unwrap()
        .json()
        .unwrap();
    assert_eq!(
        follower_members["members"].as_array().map(Vec::len),
        Some(2),
        "every node derives the same era-0 roster: {follower_members}"
    );

    // ── A join dialled at the NON-leader is refused with a leader hint ───────
    // The follower is not the leader (no election driver started → topology
    // leader leads). The conf-change gate's NotLeader path forwards to the
    // leader; since the leader has no election driver either, the join hooks are
    // unbound there and it surfaces as a non-2xx — what matters for THIS test is
    // that the membership runtime is REACHED (the era-0 roster is served and the
    // status fields are populated), not that a full conf-change commits (that
    // needs the election machinery the tier-3 suites drive).
    let join_resp = client
        .post(format!("{follower_base}/cluster/join"))
        .json(&serde_json::json!({
            "name": "ap-south",
            "grpc_addr": "ap.svc:9500",
            "http_addr": "http://ap.svc:9501"
        }))
        .send()
        .unwrap();
    assert!(
        !join_resp.status().is_success(),
        "a join with no elected leadership cannot commit a conf-change (era-0 \
         topology era): {}",
        join_resp.status()
    );

    rt.shutdown_timeout(Duration::from_secs(2));
}

/// m11p5 §3.6 precedence: a node booting with BOTH a stale durable membership
/// cache file AND a newer WAL-recovered `ClusterMembership` cell adopts the
/// CELL's roster (the cell wins) and REWRITES the cache from the cell.
///
/// We seed a persistent data dir with a v5 kind-4 record (the cell), drop the
/// seeding engine, write a STALE v3 cache claiming a different roster, then build
/// the region node on that dir. Its effective roster must reflect v5 (the cell),
/// and the on-disk cache must have been rewritten to v5.
#[test]
fn region_membership_cell_wins_over_stale_cache_and_rewrites_it() {
    use tidaldb::db::config::{NodeConfig, NodeRole};
    use tidaldb::replication::{MembershipSnapshot, MembershipStore, ShardId};

    let pair = Pair::new();
    let dir = region_dir();
    let data_dir = dir.path().to_path_buf();

    // ── (1) Seed the WAL with a v5 kind-4 record (the CELL). Roster: us-east a
    // Voter, eu-west a LEARNER (a role distinct from era-0's all-voters, so a
    // wrong source is observable). Drop the engine to release the data-dir lock.
    let v5 = MembershipRecord {
        version: 5,
        term: 3,
        members: vec![
            MemberEntry {
                id: 0,
                name: pair.leader_name.clone(),
                grpc_addr: pair.leader_grpc.to_string(),
                http_addr: pair.leader_http.to_string(),
                role: MemberRole::Voter,
            },
            MemberEntry {
                id: 1,
                name: pair.follower_name.clone(),
                grpc_addr: pair.follower_grpc.to_string(),
                http_addr: pair.follower_http.to_string(),
                role: MemberRole::Learner,
            },
        ],
    };
    {
        let db = tidaldb::TidalDb::builder()
            .with_data_dir(&data_dir)
            .with_schema(region_schema())
            .with_cluster(NodeConfig {
                role: NodeRole::Single,
                shard_id: ShardId(0),
                peer_shards: vec![ShardId(1)],
                ..NodeConfig::default()
            })
            .open()
            .expect("seed engine opens");
        db.append_membership_record(v5).expect("append v5 record");
        assert_eq!(
            db.cluster_membership().map(|(v, ..)| v),
            Some(5),
            "the cell folded v5"
        );
        // Drop releases the WAL/lock so the node can re-open the same dir.
    }

    // ── (2) Write a STALE v3 cache claiming BOTH regions are voters (an older
    // boot's roster, pre the eu-west→Learner change).
    let store = MembershipStore::new(&data_dir);
    let stale = MembershipSnapshot {
        version: 3,
        term: 2,
        members: vec![
            MemberEntry {
                id: 0,
                name: pair.leader_name.clone(),
                grpc_addr: pair.leader_grpc.to_string(),
                http_addr: pair.leader_http.to_string(),
                role: MemberRole::Voter,
            },
            MemberEntry {
                id: 1,
                name: pair.follower_name.clone(),
                grpc_addr: pair.follower_grpc.to_string(),
                http_addr: pair.follower_http.to_string(),
                role: MemberRole::Voter,
            },
        ],
    };
    store.persist(&stale).expect("write stale cache");
    assert_eq!(store.load().unwrap().map(|s| s.version), Some(3));

    // ── (3) Build the node on that dir. The view boots FROM THE CELL (v5), and
    // the boot-time reconcile rewrites the cache to v5.
    let node = build_region(pair.topology(), &pair.leader_name, &dir);

    // The membership runtime serves the CELL's roster (v5), not the stale cache.
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .unwrap();
    serve(
        &rt,
        build_region_router(Arc::new(node), mk_test_creds()),
        pair.leader_http,
    );
    let client = reqwest::blocking::Client::new();
    let members: serde_json::Value = client
        .get(format!("http://{}/cluster/members", pair.leader_http))
        .send()
        .unwrap()
        .json()
        .unwrap();
    assert_eq!(
        members["membership_version"].as_u64(),
        Some(5),
        "the cell (v5) wins over the stale cache (v3): {members}"
    );
    let roster: std::collections::HashMap<u64, String> = members["members"]
        .as_array()
        .unwrap()
        .iter()
        .map(|m| {
            (
                m["id"].as_u64().unwrap(),
                m["role"].as_str().unwrap().to_string(),
            )
        })
        .collect();
    assert_eq!(roster.get(&0).map(String::as_str), Some("voter"));
    assert_eq!(
        roster.get(&1).map(String::as_str),
        Some("learner"),
        "the cell's eu-west→Learner role wins, not the stale cache's voter"
    );

    // ── The on-disk cache was REWRITTEN from the cell (v5), correcting the stale
    // v3 (§3.6 "rewritten from the cell after every applied record" — here at the
    // boot reconcile).
    let rewritten = store.load().unwrap().expect("cache still present");
    assert_eq!(
        rewritten.version, 5,
        "the stale cache was rewritten to the cell's v5"
    );
    assert_eq!(
        rewritten.members.iter().find(|m| m.id == 1).map(|m| m.role),
        Some(MemberRole::Learner),
        "the rewritten cache carries the cell's eu-west→Learner role"
    );

    rt.shutdown_timeout(Duration::from_secs(2));
}