//! Phase 5 (ref-store, tag-store, prewarm) RPC tests split out of
//! `tests.rs` to keep both files under the 1300-line ceiling. Same
//! module namespace via `#[path]` from `rpc.rs`.

use super::*;
use crate::cluster::transport::{NodeIdentity, QuicClient, QuicServer};
use crate::config::ClusterConfig;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicU16, Ordering};
use std::time::Duration;

/// Dedicated port range for Phase-5 RPC tests. Different range from
/// tests.rs (43000+) so cross-file parallel execution can't collide.
static NEXT_PORT: AtomicU16 = AtomicU16::new(45001);
fn next_port() -> u16 {
    NEXT_PORT.fetch_add(1, Ordering::Relaxed)
}

fn loopback(port: u16) -> SocketAddr {
    format!("127.0.0.1:{port}").parse().unwrap()
}

async fn bootstrap_gossip(name: &str, port: u16) -> Arc<ClusterGossip> {
    let cfg = ClusterConfig {
        zone: "fabric-10g".into(),
        bind_lan: Some(loopback(port)),
        ..Default::default()
    };
    Arc::new(ClusterGossip::bootstrap(&cfg, name).await.unwrap())
}

async fn router_with_blobs(name: &str, port: u16) -> (tempfile::TempDir, Arc<RpcRouter>) {
    let gossip = bootstrap_gossip(name, port).await;
    let tmp = tempfile::TempDir::new().unwrap();
    let store = Arc::new(BlobStore::open(tmp.path().to_path_buf()).unwrap());
    let router = Arc::new(
        RpcRouter::new(gossip, name.into(), "fabric-10g".into()).with_blob_store(store),
    );
    (tmp, router)
}

// ── Phase 5b: reference-store RPC ────────────────────────────────────

async fn router_with_blobs_and_refs(name: &str, port: u16) -> (tempfile::TempDir, Arc<RpcRouter>) {
    use crate::cluster::refs::RefStore;
    let gossip = bootstrap_gossip(name, port).await;
    let tmp = tempfile::TempDir::new().unwrap();
    let blob_store =
        Arc::new(crate::cluster::blob::BlobStore::open(tmp.path().join("blobs")).unwrap());
    let ref_store = Arc::new(RefStore::open(tmp.path().join("refs-db")).unwrap());
    let router = Arc::new(
        RpcRouter::new(gossip, name.into(), "fabric-10g".into())
            .with_blob_store(blob_store)
            .with_ref_store(ref_store),
    );
    (tmp, router)
}

#[test]
fn phase_5b_method_byte_encoding() {
    assert_eq!(Method::GetRef.as_byte(), 0x0d);
    assert_eq!(Method::PutRef.as_byte(), 0x0e);
    assert_eq!(Method::from_byte(0x0d), Some(Method::GetRef));
    assert_eq!(Method::from_byte(0x0e), Some(Method::PutRef));
}

#[tokio::test]
async fn get_ref_returns_not_found_for_missing() {
    let (_tmp, router) = router_with_blobs_and_refs("solo", next_port()).await;
    let key = [0u8; 32];
    let mut req = vec![Method::GetRef.as_byte()];
    req.extend_from_slice(&key);
    assert_eq!(
        dispatch(&router, &req).await,
        vec![ErrorCode::NotFound.as_byte()]
    );
}

#[tokio::test]
async fn put_ref_stores_and_get_ref_reads_back() {
    let (_tmp, router) = router_with_blobs_and_refs("solo", next_port()).await;
    let key = [0x11u8; 32];
    let value = [0x22u8; 32];
    let mut put = vec![Method::PutRef.as_byte()];
    put.extend_from_slice(&key);
    put.extend_from_slice(&value);
    assert_eq!(dispatch(&router, &put).await, vec![STREAM_STATUS_OK]);

    let mut get = vec![Method::GetRef.as_byte()];
    get.extend_from_slice(&key);
    assert_eq!(dispatch(&router, &get).await, value.to_vec());
}

#[tokio::test]
async fn put_ref_rejects_wrong_length_payload() {
    let (_tmp, router) = router_with_blobs_and_refs("solo", next_port()).await;
    // 63 bytes — one shy of the 32+32 requirement.
    let req = {
        let mut r = vec![Method::PutRef.as_byte()];
        r.extend(vec![0u8; 63]);
        r
    };
    assert_eq!(
        dispatch(&router, &req).await,
        vec![ErrorCode::InvalidRequest.as_byte()]
    );
}

#[tokio::test]
async fn get_ref_rejects_wrong_length_payload() {
    let (_tmp, router) = router_with_blobs_and_refs("solo", next_port()).await;
    let req = vec![Method::GetRef.as_byte(), 0, 1, 2];
    assert_eq!(
        dispatch(&router, &req).await,
        vec![ErrorCode::InvalidRequest.as_byte()]
    );
}

#[tokio::test]
async fn ref_rpcs_return_not_configured_without_store() {
    let gossip = bootstrap_gossip("solo", next_port()).await;
    // Router with a blob store but NO ref store.
    let tmp = tempfile::TempDir::new().unwrap();
    let blob =
        Arc::new(crate::cluster::blob::BlobStore::open(tmp.path().to_path_buf()).unwrap());
    let router = Arc::new(
        RpcRouter::new(gossip, "solo".into(), "z".into()).with_blob_store(blob),
    );

    for method in [Method::GetRef, Method::PutRef] {
        let mut req = vec![method.as_byte()];
        req.extend_from_slice(&[0u8; 32]);
        req.extend_from_slice(&[0u8; 32]);
        assert_eq!(
            dispatch(&router, &req).await,
            vec![ErrorCode::NotConfigured.as_byte()],
            "method {method:?} should be NotConfigured"
        );
    }
}

#[tokio::test]
async fn end_to_end_put_ref_get_ref_over_real_quic() {
    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
    let (_tmp, router) = router_with_blobs_and_refs("a", next_port()).await;

    let server = QuicServer::bind(loopback(0), id_a).unwrap();
    let server_addr = server.local_addr().unwrap();
    let router_srv = router.clone();
    let accept_task = tokio::spawn(async move {
        if let Some(Ok(conn)) = server.accept().await {
            let _ = serve_connection(conn, router_srv).await;
        }
    });

    let client = QuicClient::new(loopback(0), id_b).unwrap();
    let conn = client.connect(server_addr, "a").await.unwrap();

    let key = [0x77u8; 32];
    let value = [0x88u8; 32];

    // Miss first.
    assert!(call_get_ref(&conn, &key).await.unwrap().is_none());
    // Put.
    call_put_ref(&conn, &key, &value).await.unwrap();
    // Hit.
    assert_eq!(call_get_ref(&conn, &key).await.unwrap(), Some(value));

    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    accept_task.abort();
}

// ── Phase 5d: tag-store RPC ──────────────────────────────────────────

async fn router_with_full_stack(name: &str, port: u16) -> (tempfile::TempDir, Arc<RpcRouter>) {
    use crate::cluster::refs::RefStore;
    use crate::cluster::tags::TagStore;
    let gossip = bootstrap_gossip(name, port).await;
    let tmp = tempfile::TempDir::new().unwrap();
    let blob_store =
        Arc::new(crate::cluster::blob::BlobStore::open(tmp.path().join("blobs")).unwrap());
    let ref_store = Arc::new(RefStore::open(tmp.path().join("refs-db")).unwrap());
    let tag_store = Arc::new(TagStore::open(tmp.path().join("tags-db")).unwrap());
    let router = Arc::new(
        RpcRouter::new(gossip, name.into(), "fabric-10g".into())
            .with_blob_store(blob_store)
            .with_ref_store(ref_store)
            .with_tag_store(tag_store),
    );
    (tmp, router)
}

#[test]
fn phase_5d_method_byte_encoding() {
    assert_eq!(Method::PutTag.as_byte(), 0x0f);
    assert_eq!(Method::GetTag.as_byte(), 0x10);
    assert_eq!(Method::DeleteTag.as_byte(), 0x11);
    assert_eq!(Method::ListTags.as_byte(), 0x12);
    for m in [Method::PutTag, Method::GetTag, Method::DeleteTag, Method::ListTags] {
        assert_eq!(Method::from_byte(m.as_byte()), Some(m));
    }
}

#[tokio::test]
async fn tag_rpcs_return_not_configured_without_store() {
    let gossip = bootstrap_gossip("solo", next_port()).await;
    let router = RpcRouter::new(gossip, "solo".into(), "z".into());
    for method in [Method::PutTag, Method::GetTag, Method::DeleteTag, Method::ListTags] {
        let mut req = vec![method.as_byte()];
        req.extend_from_slice(b"any-key");
        let reply = dispatch(&router, &req).await;
        assert_eq!(
            reply,
            vec![ErrorCode::NotConfigured.as_byte()],
            "method {method:?} should be NotConfigured"
        );
    }
}

#[tokio::test]
async fn put_tag_stores_and_get_tag_reads_back() {
    let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
    let key = "clawverse:main:latest";
    let value = [0x77u8; 32];

    let put_payload = crate::cluster::tags::encode_record(key, &value);
    let mut put_req = vec![Method::PutTag.as_byte()];
    put_req.extend_from_slice(&put_payload);
    assert_eq!(dispatch(&router, &put_req).await, vec![STREAM_STATUS_OK]);

    let mut get_req = vec![Method::GetTag.as_byte()];
    get_req.extend_from_slice(key.as_bytes());
    assert_eq!(dispatch(&router, &get_req).await, value.to_vec());
}

#[tokio::test]
async fn get_tag_returns_not_found_for_missing() {
    let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
    let mut req = vec![Method::GetTag.as_byte()];
    req.extend_from_slice(b"never-set");
    assert_eq!(
        dispatch(&router, &req).await,
        vec![ErrorCode::NotFound.as_byte()]
    );
}

#[tokio::test]
async fn get_tag_rejects_empty_key() {
    let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
    let req = vec![Method::GetTag.as_byte()]; // empty payload
    assert_eq!(
        dispatch(&router, &req).await,
        vec![ErrorCode::InvalidRequest.as_byte()]
    );
}

#[tokio::test]
async fn delete_tag_removes_and_returns_not_found_after() {
    let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
    let store = router.tag_store().unwrap().clone();
    store.put("removable", &[0u8; 32]).await.unwrap();

    let mut req = vec![Method::DeleteTag.as_byte()];
    req.extend_from_slice(b"removable");
    assert_eq!(dispatch(&router, &req).await, vec![STREAM_STATUS_OK]);

    // Second delete → NotFound.
    assert_eq!(
        dispatch(&router, &req).await,
        vec![ErrorCode::NotFound.as_byte()]
    );
}

#[tokio::test]
async fn list_tags_returns_json_sorted() {
    let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
    let store = router.tag_store().unwrap().clone();
    store.put("bravo", &[2u8; 32]).await.unwrap();
    store.put("alpha", &[1u8; 32]).await.unwrap();

    let req = vec![Method::ListTags.as_byte()];
    let reply = dispatch(&router, &req).await;
    let decoded: Vec<crate::cluster::tags::TagEntry> =
        serde_json::from_slice(&reply).unwrap();
    assert_eq!(decoded.len(), 2);
    assert_eq!(decoded[0].key, "alpha");
    assert_eq!(decoded[1].key, "bravo");
    assert_eq!(decoded[0].decode_value().unwrap(), [1u8; 32]);
}

// ── Phase 5g: cache metrics RPC ──────────────────────────────────────

#[test]
fn phase_5g_method_byte_encoding() {
    assert_eq!(Method::GetMetrics.as_byte(), 0x13);
    assert_eq!(Method::from_byte(0x13), Some(Method::GetMetrics));
}

#[tokio::test]
async fn get_metrics_returns_empty_snapshot_before_any_activity() {
    let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
    let reply = dispatch(&router, &[Method::GetMetrics.as_byte()]).await;
    let snapshot: crate::cluster::metrics::MetricsReply =
        serde_json::from_slice(&reply).unwrap();
    assert!(snapshot.started_unix > 0);
    assert_eq!(snapshot.get_ref_hits, 0);
    assert_eq!(snapshot.get_ref_misses, 0);
    assert_eq!(snapshot.blob_get_bytes, 0);
}

#[tokio::test]
async fn get_ref_records_hit_and_miss_counters() {
    let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
    let store = router.ref_store().unwrap().clone();
    let key = [0x11u8; 32];
    store.put(&key, &[0x22u8; 32]).await.unwrap();

    // Two hits.
    for _ in 0..2 {
        let mut req = vec![Method::GetRef.as_byte()];
        req.extend_from_slice(&key);
        let reply = dispatch(&router, &req).await;
        assert_eq!(reply.len(), 32);
    }
    // One miss.
    let mut req = vec![Method::GetRef.as_byte()];
    req.extend_from_slice(&[0xffu8; 32]);
    let reply = dispatch(&router, &req).await;
    assert_eq!(reply, vec![ErrorCode::NotFound.as_byte()]);

    let snapshot = router.metrics().snapshot();
    assert_eq!(snapshot.get_ref_hits, 2);
    assert_eq!(snapshot.get_ref_misses, 1);
}

#[tokio::test]
async fn get_tag_records_hit_and_miss_counters() {
    let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
    let store = router.tag_store().unwrap().clone();
    store.put("clawverse:main", &[0u8; 32]).await.unwrap();

    let mut hit = vec![Method::GetTag.as_byte()];
    hit.extend_from_slice(b"clawverse:main");
    dispatch(&router, &hit).await;

    let mut miss = vec![Method::GetTag.as_byte()];
    miss.extend_from_slice(b"never-set");
    dispatch(&router, &miss).await;

    let snapshot = router.metrics().snapshot();
    assert_eq!(snapshot.get_tag_hits, 1);
    assert_eq!(snapshot.get_tag_misses, 1);
}

#[tokio::test]
async fn blob_get_and_blob_put_record_byte_counts() {
    let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
    let payload = b"metrics witness";

    // Put — should record blob_put_bytes.
    let mut put_req = vec![Method::BlobPut.as_byte()];
    put_req.extend_from_slice(payload);
    let put_reply = dispatch(&router, &put_req).await;
    assert_eq!(put_reply.len(), 32);
    let mut id_bytes = [0u8; 32];
    id_bytes.copy_from_slice(&put_reply);

    // Get — should record blob_get_bytes.
    let mut get_req = vec![Method::BlobGet.as_byte()];
    get_req.extend_from_slice(&id_bytes);
    let get_reply = dispatch(&router, &get_req).await;
    assert_eq!(get_reply, payload);

    let snapshot = router.metrics().snapshot();
    assert_eq!(snapshot.blob_put_bytes, payload.len() as u64);
    assert_eq!(snapshot.blob_get_bytes, payload.len() as u64);
}

#[tokio::test]
async fn has_chunk_and_get_chunk_record_hit_miss_counters() {
    let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
    let store = router.blob_store().unwrap().clone();
    let bytes = b"chunk-in-store";
    let hash = crate::cluster::blob::ChunkHash::from_bytes(blake3::hash(bytes).into());
    store.put_chunk(&hash, bytes).await.unwrap();

    // HasChunk hit + miss.
    let mut hit = vec![Method::HasChunk.as_byte()];
    hit.extend_from_slice(hash.as_bytes());
    dispatch(&router, &hit).await;

    let mut miss = vec![Method::HasChunk.as_byte()];
    miss.extend_from_slice(&[0u8; 32]);
    dispatch(&router, &miss).await;

    // GetChunk hit + miss.
    let mut get_hit = vec![Method::GetChunk.as_byte()];
    get_hit.extend_from_slice(hash.as_bytes());
    dispatch(&router, &get_hit).await;

    let mut get_miss = vec![Method::GetChunk.as_byte()];
    get_miss.extend_from_slice(&[0u8; 32]);
    dispatch(&router, &get_miss).await;

    let snapshot = router.metrics().snapshot();
    assert_eq!(snapshot.has_chunk_hits, 1);
    assert_eq!(snapshot.has_chunk_misses, 1);
    assert_eq!(snapshot.get_chunk_hits, 1);
    assert_eq!(snapshot.get_chunk_misses, 1);
    // Get_chunk hit also records blob_get_bytes.
    assert_eq!(snapshot.blob_get_bytes, bytes.len() as u64);
}

#[tokio::test]
async fn end_to_end_get_metrics_over_real_quic() {
    // Exercise every counter, then fetch the metrics reply through
    // real QUIC and verify each field.
    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
    let (_tmp, router) = router_with_full_stack("a", next_port()).await;

    // Seed some activity locally so the counters have real values.
    let store = router.ref_store().unwrap().clone();
    store.put(&[0x11u8; 32], &[0x22u8; 32]).await.unwrap();
    let tag_store = router.tag_store().unwrap().clone();
    tag_store.put("hit-me", &[0u8; 32]).await.unwrap();
    let blob_store = router.blob_store().unwrap().clone();
    let payload = vec![0x5au8; 4096];
    let blob_id = blob_store.put_bytes(&payload).await.unwrap();

    // Fire dispatches to move the counters.
    let mut ref_hit = vec![Method::GetRef.as_byte()];
    ref_hit.extend_from_slice(&[0x11u8; 32]);
    dispatch(&router, &ref_hit).await;

    let mut ref_miss = vec![Method::GetRef.as_byte()];
    ref_miss.extend_from_slice(&[0x99u8; 32]);
    dispatch(&router, &ref_miss).await;

    let mut tag_hit = vec![Method::GetTag.as_byte()];
    tag_hit.extend_from_slice(b"hit-me");
    dispatch(&router, &tag_hit).await;

    let mut blob_get = vec![Method::BlobGet.as_byte()];
    blob_get.extend_from_slice(blob_id.as_bytes());
    dispatch(&router, &blob_get).await;

    // Now start the server + fetch metrics over the wire.
    let server = QuicServer::bind(loopback(0), id_a).unwrap();
    let server_addr = server.local_addr().unwrap();
    let router_srv = router.clone();
    let accept_task = tokio::spawn(async move {
        if let Some(Ok(conn)) = server.accept().await {
            let _ = serve_connection(conn, router_srv).await;
        }
    });

    let client = QuicClient::new(loopback(0), id_b).unwrap();
    let conn = client.connect(server_addr, "a").await.unwrap();

    let m = call_get_metrics(&conn).await.unwrap();
    assert_eq!(m.get_ref_hits, 1);
    assert_eq!(m.get_ref_misses, 1);
    assert_eq!(m.get_tag_hits, 1);
    assert_eq!(m.get_tag_misses, 0);
    assert_eq!(m.blob_get_bytes, payload.len() as u64);
    assert_eq!(m.get_ref_hit_rate(), Some(0.5));

    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    accept_task.abort();
}

#[tokio::test]
async fn end_to_end_prewarm_copies_tagged_blob_between_two_peers() {
    // Phase 5f: `claw-cargo prewarm --from A --to C --pin tag`.
    // Two full RPC servers running in-process (upstream = A, downstream
    // = C). Client is B (uses a third distinct leaf cert), fetches from
    // A, uploads to C, republishes the tag on C. Verifies the tag +
    // blob are queryable on C after the copy.
    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
    // Second pair for C. The `_id_b2` here is unused but has to share
    // the same CA as A so client B can talk to both. Since our test
    // helper generates a fresh CA per pair, we cheat by using the same
    // pair generator with distinct names — in a real deployment both
    // servers would share the fleet CA that signed B.
    let (id_c, id_b_for_c) =
        NodeIdentity::generate_test_pair("c", "b_client_for_c").unwrap();

    let (_tmp_a, router_a) = router_with_full_stack("a", next_port()).await;
    let (_tmp_c, router_c) = router_with_full_stack("c", next_port()).await;

    // Seed a blob + tag on A.
    let payload: Vec<u8> = (0..1_536_000).map(|i| (i % 251) as u8).collect();
    let blob_id = router_a
        .blob_store()
        .unwrap()
        .put_bytes(&payload)
        .await
        .unwrap();
    router_a
        .tag_store()
        .unwrap()
        .put("clawverse:main:latest", blob_id.as_bytes())
        .await
        .unwrap();

    let server_a = QuicServer::bind(loopback(0), id_a).unwrap();
    let server_a_addr = server_a.local_addr().unwrap();
    let router_a_srv = router_a.clone();
    let accept_a = tokio::spawn(async move {
        if let Some(Ok(conn)) = server_a.accept().await {
            let _ = serve_connection(conn, router_a_srv).await;
        }
    });

    let server_c = QuicServer::bind(loopback(0), id_c).unwrap();
    let server_c_addr = server_c.local_addr().unwrap();
    let router_c_srv = router_c.clone();
    let accept_c = tokio::spawn(async move {
        if let Some(Ok(conn)) = server_c.accept().await {
            let _ = serve_connection(conn, router_c_srv).await;
        }
    });

    // Client → upstream A (via B's identity).
    let up_client = QuicClient::new(loopback(0), id_b).unwrap();
    let up_conn = up_client.connect(server_a_addr, "a").await.unwrap();
    let tag_value = call_get_tag(&up_conn, "clawverse:main:latest")
        .await
        .unwrap()
        .expect("tag on A");
    let up_blob_id = BlobId::from_bytes(tag_value);
    assert_eq!(up_blob_id, blob_id);
    let mut buf: Vec<u8> = Vec::new();
    let ok = call_blob_get_stream(&up_conn, &up_blob_id, &mut buf)
        .await
        .unwrap();
    assert!(ok);
    assert_eq!(buf, payload);
    up_conn.close(quinn::VarInt::from_u32(0), b"done");
    up_client.shutdown().await;

    // Client → downstream C. Upload the bytes + publish the tag.
    let down_client = QuicClient::new(loopback(0), id_b_for_c).unwrap();
    let down_conn = down_client.connect(server_c_addr, "c").await.unwrap();
    let cursor = std::io::Cursor::new(buf.clone());
    let assigned = call_blob_put_stream(&down_conn, cursor).await.unwrap();
    assert_eq!(assigned, blob_id, "content-addressed → same BlobId");
    call_put_tag(&down_conn, "clawverse:main:latest", blob_id.as_bytes())
        .await
        .unwrap();
    down_conn.close(quinn::VarInt::from_u32(0), b"done");
    down_client.shutdown().await;

    // Give the accept tasks a moment, then verify C ended up with both
    // the blob AND the tag — the two things a subsequent
    // `prefetch --pin` would look for.
    tokio::time::sleep(Duration::from_millis(50)).await;
    let c_blob = router_c
        .blob_store()
        .unwrap()
        .get_bytes(&blob_id)
        .await
        .unwrap();
    assert_eq!(c_blob.as_deref(), Some(payload.as_slice()));
    let c_tag = router_c
        .tag_store()
        .unwrap()
        .get("clawverse:main:latest")
        .await
        .unwrap();
    assert_eq!(c_tag, Some(*blob_id.as_bytes()));

    accept_a.abort();
    accept_c.abort();
}

#[tokio::test]
async fn end_to_end_tag_resolve_and_stream_restore_over_real_quic() {
    // Phase 5e: the pipeline `claw-cargo prefetch --pin <tag>` runs
    // internally — put a blob, publish a tag pointing at it, then
    // GetTag → BlobStat → BlobGetStream to reassemble the content
    // byte-equal on the other side.
    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
    let (_tmp, router) = router_with_full_stack("a", next_port()).await;

    // Seed the blob store with a synthetic "captured target dir".
    let payload: Vec<u8> = (0..2 * 1024 * 1024).map(|i| (i % 251) as u8).collect();
    let blob_id = router
        .blob_store()
        .unwrap()
        .put_bytes(&payload)
        .await
        .unwrap();
    // Publish a tag pointing at that blob.
    router
        .tag_store()
        .unwrap()
        .put("clawverse:main:latest", blob_id.as_bytes())
        .await
        .unwrap();

    let server = QuicServer::bind(loopback(0), id_a).unwrap();
    let server_addr = server.local_addr().unwrap();
    let router_srv = router.clone();
    let accept_task = tokio::spawn(async move {
        if let Some(Ok(conn)) = server.accept().await {
            let _ = serve_connection(conn, router_srv).await;
        }
    });

    let client = QuicClient::new(loopback(0), id_b).unwrap();
    let conn = client.connect(server_addr, "a").await.unwrap();

    // Client side of `prefetch --pin <tag>`:
    //   1. GetTag → BlobId bytes
    //   2. BlobStat → confirm blob exists + size
    //   3. BlobGetStream → download into buffer
    let tag_value = call_get_tag(&conn, "clawverse:main:latest")
        .await
        .unwrap()
        .expect("tag set");
    assert_eq!(tag_value, *blob_id.as_bytes());
    let resolved_id = BlobId::from_bytes(tag_value);
    let stat = call_blob_stat(&conn, &resolved_id).await.unwrap().unwrap();
    assert_eq!(stat.total_size, payload.len() as u64);

    let mut sink: Vec<u8> = Vec::new();
    let ok = call_blob_get_stream(&conn, &resolved_id, &mut sink)
        .await
        .unwrap();
    assert!(ok);
    assert_eq!(sink.len(), payload.len());
    assert_eq!(sink, payload, "reassembled bytes match source");

    // Missing-tag path: prefetch --pin never-set-name reports None.
    let missing = call_get_tag(&conn, "never-set").await.unwrap();
    assert!(missing.is_none());

    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    accept_task.abort();
}

#[tokio::test]
async fn end_to_end_pin_lookup_delete_over_real_quic() {
    // Full flow: publish a tag → look it up → list → delete → confirm gone.
    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
    let (_tmp, router) = router_with_full_stack("a", next_port()).await;

    let server = QuicServer::bind(loopback(0), id_a).unwrap();
    let server_addr = server.local_addr().unwrap();
    let router_srv = router.clone();
    let accept_task = tokio::spawn(async move {
        if let Some(Ok(conn)) = server.accept().await {
            let _ = serve_connection(conn, router_srv).await;
        }
    });

    let client = QuicClient::new(loopback(0), id_b).unwrap();
    let conn = client.connect(server_addr, "a").await.unwrap();

    let key = "clawverse:main:latest-cache";
    let value = [0xaau8; 32];

    // Miss first.
    assert!(call_get_tag(&conn, key).await.unwrap().is_none());
    // Publish.
    call_put_tag(&conn, key, &value).await.unwrap();
    // Hit.
    assert_eq!(call_get_tag(&conn, key).await.unwrap(), Some(value));
    // List sees it.
    let list = call_list_tags(&conn).await.unwrap();
    assert_eq!(list.len(), 1);
    assert_eq!(list[0].key, key);
    // Delete.
    assert!(call_delete_tag(&conn, key).await.unwrap());
    // Gone.
    assert!(call_get_tag(&conn, key).await.unwrap().is_none());
    assert!(!call_delete_tag(&conn, key).await.unwrap()); // second delete
    assert!(call_list_tags(&conn).await.unwrap().is_empty());

    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    accept_task.abort();
}

#[tokio::test]
async fn call_get_chunk_verifies_returned_hash() {
    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
    let (_tmp, router) = router_with_blobs("a", next_port()).await;
    let bytes = b"chunk to fetch";
    let hash = ChunkHash::from_bytes(blake3::hash(bytes).into());
    router
        .blob_store()
        .unwrap()
        .put_chunk(&hash, bytes)
        .await
        .unwrap();

    let server = QuicServer::bind(loopback(0), id_a).unwrap();
    let server_addr = server.local_addr().unwrap();
    let router_srv = router.clone();
    let accept_task = tokio::spawn(async move {
        if let Some(Ok(conn)) = server.accept().await {
            let _ = serve_connection(conn, router_srv).await;
        }
    });

    let client = QuicClient::new(loopback(0), id_b).unwrap();
    let conn = client.connect(server_addr, "a").await.unwrap();
    let got = call_get_chunk(&conn, &hash).await.unwrap().unwrap();
    assert_eq!(got, bytes);

    let missing = ChunkHash::from_bytes([0u8; 32]);
    assert!(call_get_chunk(&conn, &missing).await.unwrap().is_none());

    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    accept_task.abort();
}

#[tokio::test]
async fn end_to_end_stream_put_and_get_over_real_quic() {
    // The whole point of Phase 2c: put a big blob without holding
    // it in memory on either side. This test drives a 12 MiB
    // payload (3 chunks) through BlobPutStream and BlobGetStream.
    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
    let (_tmp, router) = router_with_blobs("a", next_port()).await;

    let server = QuicServer::bind(loopback(0), id_a).unwrap();
    let server_addr = server.local_addr().unwrap();
    let router_srv = router.clone();
    let accept_task = tokio::spawn(async move {
        if let Some(Ok(conn)) = server.accept().await {
            let _ = serve_connection(conn, router_srv).await;
        }
    });

    let client = QuicClient::new(loopback(0), id_b).unwrap();
    let conn = client.connect(server_addr, "a").await.unwrap();

    // 12 MiB payload; deliberately not a chunk multiple so the last
    // chunk is short.
    let mut payload: Vec<u8> = Vec::with_capacity(12 * 1024 * 1024 + 777);
    for i in 0..12 * 1024 * 1024 + 777 {
        payload.push((i % 251) as u8);
    }
    let reader = std::io::Cursor::new(payload.clone());
    let id = call_blob_put_stream(&conn, reader).await.unwrap();
    let expected = BlobId::from_bytes(blake3::hash(&payload).into());
    assert_eq!(id, expected);

    // Manifest verifies the chunk split.
    let manifest = call_blob_load_manifest(&conn, &id).await.unwrap().unwrap();
    assert_eq!(manifest.chunks.len(), 4, "12 MiB + 777 → 4 chunks");
    assert_eq!(manifest.total_size, payload.len() as u64);

    // Stream it back.
    let mut sink: Vec<u8> = Vec::new();
    let ok = call_blob_get_stream(&conn, &id, &mut sink).await.unwrap();
    assert!(ok);
    assert_eq!(sink.len(), payload.len());
    assert_eq!(sink, payload);

    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    accept_task.abort();
}

#[tokio::test]
async fn stream_get_returns_false_for_missing_blob() {
    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
    let (_tmp, router) = router_with_blobs("a", next_port()).await;
    let server = QuicServer::bind(loopback(0), id_a).unwrap();
    let server_addr = server.local_addr().unwrap();
    let router_srv = router.clone();
    let accept_task = tokio::spawn(async move {
        if let Some(Ok(conn)) = server.accept().await {
            let _ = serve_connection(conn, router_srv).await;
        }
    });

    let client = QuicClient::new(loopback(0), id_b).unwrap();
    let conn = client.connect(server_addr, "a").await.unwrap();

    let ghost = BlobId::from_bytes([0u8; 32]);
    let mut sink: Vec<u8> = Vec::new();
    let ok = call_blob_get_stream(&conn, &ghost, &mut sink)
        .await
        .unwrap();
    assert!(!ok);
    assert!(sink.is_empty());

    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    accept_task.abort();
}

#[tokio::test]
async fn stream_methods_return_not_configured_without_store() {
    // Router built without a store: streaming methods must reply
    // with NotConfigured as their first status byte.
    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
    let gossip = bootstrap_gossip("a", next_port()).await;
    // NOTE: no `.with_blob_store(...)` — Blob* methods should error.
    let router = Arc::new(RpcRouter::new(gossip, "a".into(), "z".into()));

    let server = QuicServer::bind(loopback(0), id_a).unwrap();
    let server_addr = server.local_addr().unwrap();
    let router_srv = router.clone();
    let accept_task = tokio::spawn(async move {
        if let Some(Ok(conn)) = server.accept().await {
            let _ = serve_connection(conn, router_srv).await;
        }
    });

    let client = QuicClient::new(loopback(0), id_b).unwrap();
    let conn = client.connect(server_addr, "a").await.unwrap();

    // BlobPutStream on a store-less router → NotConfigured.
    let reader = std::io::Cursor::new(b"never lands".to_vec());
    let put_err = call_blob_put_stream(&conn, reader)
        .await
        .err()
        .expect("no store → error");
    assert!(
        put_err.to_string().contains("not configured"),
        "put_stream err: {put_err}"
    );

    // BlobGetStream: same error surface.
    let ghost = BlobId::from_bytes([0u8; 32]);
    let mut sink: Vec<u8> = Vec::new();
    let get_err = call_blob_get_stream(&conn, &ghost, &mut sink)
        .await
        .err()
        .expect("no store → error");
    assert!(
        get_err.to_string().contains("not configured"),
        "get_stream err: {get_err}"
    );

    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    accept_task.abort();
}

#[tokio::test]
async fn stream_put_deduplicates_with_prior_put_bytes() {
    // Uploading the same content twice — once bounded, once
    // streaming — yields the same BlobId AND doesn't double-store
    // chunks. Proves stream + bounded are consistent addresses.
    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
    let (_tmp, router) = router_with_blobs("a", next_port()).await;

    // Pre-populate via the local store's bounded API.
    let payload = vec![0xdeu8; 4 * 1024 * 1024 + 100];
    let pre_id = router
        .blob_store()
        .unwrap()
        .put_bytes(&payload)
        .await
        .unwrap();

    let server = QuicServer::bind(loopback(0), id_a).unwrap();
    let server_addr = server.local_addr().unwrap();
    let router_srv = router.clone();
    let accept_task = tokio::spawn(async move {
        if let Some(Ok(conn)) = server.accept().await {
            let _ = serve_connection(conn, router_srv).await;
        }
    });
    let client = QuicClient::new(loopback(0), id_b).unwrap();
    let conn = client.connect(server_addr, "a").await.unwrap();

    let reader = std::io::Cursor::new(payload.clone());
    let stream_id = call_blob_put_stream(&conn, reader).await.unwrap();
    assert_eq!(pre_id, stream_id);

    // The manifest is still there and its chunk count matches the
    // pre-existing one — no fork.
    let manifest = router
        .blob_store()
        .unwrap()
        .load_manifest(&stream_id)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(manifest.chunks.len(), 2);

    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    accept_task.abort();
}

#[tokio::test]
async fn end_to_end_streaming_prewarm_copies_chunks_bounded_memory() {
    // Phase 5h: `prewarm_missing_chunks_between` streams chunks one at
    // a time from upstream to downstream without holding the whole
    // blob in RAM. Two full RPC servers; seed a multi-chunk blob on A;
    // stream it to C; verify downstream ended up with every chunk +
    // manifest and can serve the assembled bytes back.
    use crate::cluster::blob::CHUNK_SIZE;

    let (id_a, id_b_for_a) = NodeIdentity::generate_test_pair("a", "b_a").unwrap();
    let (id_c, id_b_for_c) = NodeIdentity::generate_test_pair("c", "b_c").unwrap();

    let (_tmp_a, router_a) = router_with_full_stack("a", next_port()).await;
    let (_tmp_c, router_c) = router_with_full_stack("c", next_port()).await;

    // 3 chunks worth so we're not testing a single-chunk edge case.
    // The last chunk is intentionally short (not a full CHUNK_SIZE) so
    // we also cover the tail-chunk path.
    let payload: Vec<u8> = (0..(2 * CHUNK_SIZE + CHUNK_SIZE / 4))
        .map(|i| ((i * 31) % 251) as u8)
        .collect();
    let blob_id = router_a
        .blob_store()
        .unwrap()
        .put_bytes(&payload)
        .await
        .unwrap();
    let up_manifest = router_a
        .blob_store()
        .unwrap()
        .load_manifest(&blob_id)
        .await
        .unwrap()
        .expect("manifest on A");
    assert_eq!(up_manifest.chunks.len(), 3, "expected 3 chunks in test payload");

    let server_a = QuicServer::bind(loopback(0), id_a).unwrap();
    let server_a_addr = server_a.local_addr().unwrap();
    let router_a_srv = router_a.clone();
    let accept_a = tokio::spawn(async move {
        while let Some(Ok(conn)) = server_a.accept().await {
            let r = router_a_srv.clone();
            tokio::spawn(async move {
                let _ = serve_connection(conn, r).await;
            });
        }
    });

    let server_c = QuicServer::bind(loopback(0), id_c).unwrap();
    let server_c_addr = server_c.local_addr().unwrap();
    let router_c_srv = router_c.clone();
    let accept_c = tokio::spawn(async move {
        while let Some(Ok(conn)) = server_c.accept().await {
            let r = router_c_srv.clone();
            tokio::spawn(async move {
                let _ = serve_connection(conn, r).await;
            });
        }
    });

    // Mediator client — one QUIC client per side (distinct identities
    // are needed since each test pair generates its own CA).
    let up_client = QuicClient::new(loopback(0), id_b_for_a).unwrap();
    let up_conn = up_client.connect(server_a_addr, "a").await.unwrap();
    let down_client = QuicClient::new(loopback(0), id_b_for_c).unwrap();
    let down_conn = down_client.connect(server_c_addr, "c").await.unwrap();

    let (uploaded, total) =
        prewarm_missing_chunks_between(&up_conn, &down_conn, &blob_id)
            .await
            .unwrap();
    assert_eq!(total, 3);
    assert_eq!(uploaded, 3, "cold downstream must receive every chunk");

    // Downstream reassembly proves the manifest committed AND all
    // chunks landed correctly.
    let round = router_c
        .blob_store()
        .unwrap()
        .get_bytes(&blob_id)
        .await
        .unwrap();
    assert_eq!(round.as_deref(), Some(payload.as_slice()));

    up_conn.close(quinn::VarInt::from_u32(0), b"done");
    down_conn.close(quinn::VarInt::from_u32(0), b"done");
    up_client.shutdown().await;
    down_client.shutdown().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    accept_a.abort();
    accept_c.abort();
}

#[tokio::test]
async fn streaming_prewarm_skips_chunks_already_present_downstream() {
    // Phase 5h: verify the dedup path. Pre-seed a subset of the
    // upstream blob's chunks on downstream; run the streaming prewarm;
    // `uploaded` must be less than `total` by exactly the pre-seeded
    // count.
    use crate::cluster::blob::CHUNK_SIZE;

    let (id_a, id_b_for_a) = NodeIdentity::generate_test_pair("a", "b_a").unwrap();
    let (id_c, id_b_for_c) = NodeIdentity::generate_test_pair("c", "b_c").unwrap();

    let (_tmp_a, router_a) = router_with_full_stack("a", next_port()).await;
    let (_tmp_c, router_c) = router_with_full_stack("c", next_port()).await;

    let payload: Vec<u8> = (0..(3 * CHUNK_SIZE)).map(|i| ((i * 17) % 251) as u8).collect();
    let blob_id = router_a
        .blob_store()
        .unwrap()
        .put_bytes(&payload)
        .await
        .unwrap();
    let up_manifest = router_a
        .blob_store()
        .unwrap()
        .load_manifest(&blob_id)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(up_manifest.chunks.len(), 3);

    // Pre-seed the first chunk on downstream so it's genuinely already
    // present. Reads it from A's store to guarantee identical bytes.
    let first_hash = up_manifest.chunks[0];
    let first_bytes = router_a
        .blob_store()
        .unwrap()
        .read_chunk(&first_hash)
        .await
        .unwrap()
        .unwrap();
    router_c
        .blob_store()
        .unwrap()
        .put_chunk(&first_hash, &first_bytes)
        .await
        .unwrap();

    let server_a = QuicServer::bind(loopback(0), id_a).unwrap();
    let addr_a = server_a.local_addr().unwrap();
    let ra = router_a.clone();
    let acc_a = tokio::spawn(async move {
        while let Some(Ok(conn)) = server_a.accept().await {
            let r = ra.clone();
            tokio::spawn(async move {
                let _ = serve_connection(conn, r).await;
            });
        }
    });
    let server_c = QuicServer::bind(loopback(0), id_c).unwrap();
    let addr_c = server_c.local_addr().unwrap();
    let rc = router_c.clone();
    let acc_c = tokio::spawn(async move {
        while let Some(Ok(conn)) = server_c.accept().await {
            let r = rc.clone();
            tokio::spawn(async move {
                let _ = serve_connection(conn, r).await;
            });
        }
    });

    let up_client = QuicClient::new(loopback(0), id_b_for_a).unwrap();
    let up_conn = up_client.connect(addr_a, "a").await.unwrap();
    let down_client = QuicClient::new(loopback(0), id_b_for_c).unwrap();
    let down_conn = down_client.connect(addr_c, "c").await.unwrap();

    let (uploaded, total) =
        prewarm_missing_chunks_between(&up_conn, &down_conn, &blob_id)
            .await
            .unwrap();
    assert_eq!(total, 3);
    assert_eq!(
        uploaded, 2,
        "dedup should skip the pre-seeded first chunk (uploaded={uploaded})"
    );

    // Idempotent: rerun should upload zero chunks (all present now).
    let (uploaded2, _) =
        prewarm_missing_chunks_between(&up_conn, &down_conn, &blob_id)
            .await
            .unwrap();
    assert_eq!(uploaded2, 0, "second prewarm should be a full-dedup no-op");

    let round = router_c
        .blob_store()
        .unwrap()
        .get_bytes(&blob_id)
        .await
        .unwrap();
    assert_eq!(round.as_deref(), Some(payload.as_slice()));

    up_conn.close(quinn::VarInt::from_u32(0), b"done");
    down_conn.close(quinn::VarInt::from_u32(0), b"done");
    up_client.shutdown().await;
    down_client.shutdown().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    acc_a.abort();
    acc_c.abort();
}

#[tokio::test]
async fn end_to_end_parallel_prewarm_copies_chunks_and_matches_sequential() {
    // Phase 5k: parallel prewarm variant produces the same downstream
    // state as the sequential path. Also proves the JoinSet fanout
    // doesn't lose or duplicate chunks.
    use crate::cluster::blob::CHUNK_SIZE;

    let (id_a, id_b_for_a) = NodeIdentity::generate_test_pair("a", "b_a").unwrap();
    let (id_c, id_b_for_c) = NodeIdentity::generate_test_pair("c", "b_c").unwrap();

    let (_tmp_a, router_a) = router_with_full_stack("a", next_port()).await;
    let (_tmp_c, router_c) = router_with_full_stack("c", next_port()).await;

    // 5 chunks so parallelism (concurrency=3) actually queues work.
    let payload: Vec<u8> = (0..(4 * CHUNK_SIZE + CHUNK_SIZE / 3))
        .map(|i| ((i * 7) % 251) as u8)
        .collect();
    let blob_id = router_a
        .blob_store()
        .unwrap()
        .put_bytes(&payload)
        .await
        .unwrap();
    let manifest = router_a
        .blob_store()
        .unwrap()
        .load_manifest(&blob_id)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(manifest.chunks.len(), 5, "expected 5 chunks in payload");

    let server_a = QuicServer::bind(loopback(0), id_a).unwrap();
    let addr_a = server_a.local_addr().unwrap();
    let ra = router_a.clone();
    let acc_a = tokio::spawn(async move {
        while let Some(Ok(conn)) = server_a.accept().await {
            let r = ra.clone();
            tokio::spawn(async move {
                let _ = serve_connection(conn, r).await;
            });
        }
    });
    let server_c = QuicServer::bind(loopback(0), id_c).unwrap();
    let addr_c = server_c.local_addr().unwrap();
    let rc = router_c.clone();
    let acc_c = tokio::spawn(async move {
        while let Some(Ok(conn)) = server_c.accept().await {
            let r = rc.clone();
            tokio::spawn(async move {
                let _ = serve_connection(conn, r).await;
            });
        }
    });

    let up_client = QuicClient::new(loopback(0), id_b_for_a).unwrap();
    let up_conn = up_client.connect(addr_a, "a").await.unwrap();
    let down_client = QuicClient::new(loopback(0), id_b_for_c).unwrap();
    let down_conn = down_client.connect(addr_c, "c").await.unwrap();

    let (uploaded, total) =
        prewarm_missing_chunks_between_parallel(&up_conn, &down_conn, &blob_id, 3)
            .await
            .unwrap();
    assert_eq!(total, 5);
    assert_eq!(uploaded, 5, "cold downstream must receive every chunk");

    // Round-trip proves manifest committed AND chunks landed.
    let round = router_c
        .blob_store()
        .unwrap()
        .get_bytes(&blob_id)
        .await
        .unwrap();
    assert_eq!(round.as_deref(), Some(payload.as_slice()));

    // Idempotency: re-running with concurrency=8 uploads 0 (full dedup).
    let (uploaded2, _) =
        prewarm_missing_chunks_between_parallel(&up_conn, &down_conn, &blob_id, 8)
            .await
            .unwrap();
    assert_eq!(uploaded2, 0, "re-run should be a no-op via has_chunk dedup");

    // concurrency=0 falls through to sequential.
    let (uploaded3, _) =
        prewarm_missing_chunks_between_parallel(&up_conn, &down_conn, &blob_id, 0)
            .await
            .unwrap();
    assert_eq!(uploaded3, 0, "sequential fallback should also see full dedup");

    up_conn.close(quinn::VarInt::from_u32(0), b"done");
    down_conn.close(quinn::VarInt::from_u32(0), b"done");
    up_client.shutdown().await;
    down_client.shutdown().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    acc_a.abort();
    acc_c.abort();
}

#[tokio::test]
async fn parallel_blob_get_reassembles_multi_chunk_blob_byte_equal() {
    // Field finding 2026-07-12: parallel chunk fetch on restore.
    // Verifies (1) reassembly byte-equals a sequential BlobGetStream,
    // (2) tail chunks (not full CHUNK_SIZE) land at the right offset,
    // (3) NotFound path returns None.
    use crate::cluster::blob::CHUNK_SIZE;

    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
    let (_tmp_a, router_a) = router_with_full_stack("a", next_port()).await;

    let payload: Vec<u8> = (0..(3 * CHUNK_SIZE + CHUNK_SIZE / 5))
        .map(|i| ((i * 13) % 251) as u8)
        .collect();
    let blob_id = router_a
        .blob_store()
        .unwrap()
        .put_bytes(&payload)
        .await
        .unwrap();

    let server_a = QuicServer::bind(loopback(0), id_a).unwrap();
    let addr_a = server_a.local_addr().unwrap();
    let ra = router_a.clone();
    let acc_a = tokio::spawn(async move {
        while let Some(Ok(conn)) = server_a.accept().await {
            let r = ra.clone();
            tokio::spawn(async move {
                let _ = serve_connection(conn, r).await;
            });
        }
    });

    let client = QuicClient::new(loopback(0), id_b).unwrap();
    let conn = client.connect(addr_a, "a").await.unwrap();

    // Sequential reference: BlobGetStream.
    let mut seq = Vec::new();
    let ok = call_blob_get_stream(&conn, &blob_id, &mut seq).await.unwrap();
    assert!(ok);
    assert_eq!(seq, payload, "sequential fetch must be byte-equal to source");

    // Parallel with concurrency = 4.
    let par = call_blob_get_parallel(&conn, &blob_id, 4)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(par, payload, "parallel fetch must match sequential");
    assert_eq!(par, seq, "parallel and sequential must agree");

    // concurrency = 1 falls through to still-parallel (with just one
    // in-flight) but must still be correct.
    let ser_via_par = call_blob_get_parallel(&conn, &blob_id, 1)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(ser_via_par, payload);

    // Unknown blob → None.
    let missing = crate::cluster::blob::BlobId::from_bytes([0u8; 32]);
    let none = call_blob_get_parallel(&conn, &missing, 4).await.unwrap();
    assert!(none.is_none(), "NotFound must surface as None");

    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    acc_a.abort();
}

// ── Phase 3: stamped-ref RPC end-to-end ──────────────────────────────

#[tokio::test]
async fn end_to_end_put_ref_versioned_merges_and_rejects() {
    // Two writers publish stamped refs for the same key. Higher
    // (clock, node) wins; lower is rejected; equal is idempotent.
    use crate::cluster::refs::{node_stamp_for, StampedRef};

    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
    let (_tmp_a, router_a) = router_with_blobs_and_refs("a", next_port()).await;

    let server_a = QuicServer::bind(loopback(0), id_a).unwrap();
    let addr = server_a.local_addr().unwrap();
    let ra = router_a.clone();
    let acc = tokio::spawn(async move {
        while let Some(Ok(conn)) = server_a.accept().await {
            let r = ra.clone();
            tokio::spawn(async move {
                let _ = serve_connection(conn, r).await;
            });
        }
    });
    let client = QuicClient::new(loopback(0), id_b).unwrap();
    let conn = client.connect(addr, "a").await.unwrap();

    let key = [0x77; 32];
    let node_x = node_stamp_for("runner-x");
    let node_y = node_stamp_for("runner-y");

    // Empty: nothing to return.
    assert_eq!(call_get_ref_versioned(&conn, &key).await.unwrap(), None);

    // First write: merged.
    let first = StampedRef {
        value: [0xA1; 32],
        clock: 10,
        node: node_x,
    };
    assert!(call_put_ref_versioned(&conn, &key, &first).await.unwrap());
    assert_eq!(
        call_get_ref_versioned(&conn, &key).await.unwrap(),
        Some(first)
    );

    // Older clock from a different writer: rejected.
    let older = StampedRef {
        value: [0xA2; 32],
        clock: 9,
        node: node_y,
    };
    assert!(!call_put_ref_versioned(&conn, &key, &older).await.unwrap());
    assert_eq!(
        call_get_ref_versioned(&conn, &key).await.unwrap(),
        Some(first),
        "older write must not overwrite"
    );

    // Same clock, higher node stamp: merged if node_y > node_x, else rejected.
    let tied = StampedRef {
        value: [0xA3; 32],
        clock: 10,
        node: node_y,
    };
    let merged = call_put_ref_versioned(&conn, &key, &tied).await.unwrap();
    let after_tie = call_get_ref_versioned(&conn, &key).await.unwrap().unwrap();
    if node_y > node_x {
        assert!(merged, "higher node stamp should merge");
        assert_eq!(after_tie, tied);
    } else {
        assert!(!merged, "lower node stamp should be rejected");
        assert_eq!(after_tie, first);
    }

    // Higher clock always wins.
    let latest = StampedRef {
        value: [0xA4; 32],
        clock: 100,
        node: node_x,
    };
    assert!(call_put_ref_versioned(&conn, &key, &latest).await.unwrap());
    assert_eq!(
        call_get_ref_versioned(&conn, &key).await.unwrap(),
        Some(latest)
    );

    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    acc.abort();
}

#[tokio::test]
async fn end_to_end_put_tag_versioned_merges_and_rejects() {
    // Phase 3c end-to-end: two writers publish stamped tags for the
    // same key. Higher (clock, node) wins.
    use crate::cluster::refs::node_stamp_for;
    use crate::cluster::tags::StampedTagValue;

    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
    let (_tmp_a, router_a) = router_with_full_stack("a", next_port()).await;

    let server_a = QuicServer::bind(loopback(0), id_a).unwrap();
    let addr = server_a.local_addr().unwrap();
    let ra = router_a.clone();
    let acc = tokio::spawn(async move {
        while let Some(Ok(conn)) = server_a.accept().await {
            let r = ra.clone();
            tokio::spawn(async move {
                let _ = serve_connection(conn, r).await;
            });
        }
    });
    let client = QuicClient::new(loopback(0), id_b).unwrap();
    let conn = client.connect(addr, "a").await.unwrap();

    let key = "clawverse:main:latest";
    let node_x = node_stamp_for("runner-x");
    let node_y = node_stamp_for("runner-y");

    assert_eq!(call_get_tag_versioned(&conn, key).await.unwrap(), None);

    let first = StampedTagValue {
        value: [0x11; 32],
        clock: 10,
        node: node_x,
    };
    assert!(call_put_tag_versioned(&conn, key, &first).await.unwrap());
    assert_eq!(
        call_get_tag_versioned(&conn, key).await.unwrap(),
        Some(first)
    );

    // Older clock → rejected.
    let older = StampedTagValue {
        value: [0x22; 32],
        clock: 9,
        node: node_y,
    };
    assert!(!call_put_tag_versioned(&conn, key, &older).await.unwrap());
    assert_eq!(
        call_get_tag_versioned(&conn, key).await.unwrap(),
        Some(first)
    );

    // Higher clock always merges.
    let latest = StampedTagValue {
        value: [0x33; 32],
        clock: 100,
        node: node_x,
    };
    assert!(call_put_tag_versioned(&conn, key, &latest).await.unwrap());
    assert_eq!(
        call_get_tag_versioned(&conn, key).await.unwrap(),
        Some(latest)
    );

    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;
    tokio::time::sleep(Duration::from_millis(50)).await;
    acc.abort();
}