claw-store/src/claw_cargo.rs

@@ -87,6 +87,39 @@ enum Cmd {
     Unpin(UnpinArgs),
     /// List every tag published on the peer.
     ListTags(PeerArgs),
+    /// Copy a tagged cache from one peer (upstream) to another
+    /// (downstream). Streams the blob upstream → downstream via
+    /// BlobGetStream + BlobPutStream and re-publishes the tag on the
+    /// downstream side. Useful as a Gitea-webhook target so the runner's
+    /// local daemon has the cache warm before the build starts.
+    Prewarm(PrewarmArgs),
+}
+
+#[derive(clap::Args, Debug, Clone)]
+struct PrewarmArgs {
+    /// Upstream peer name (SAN on their cert).
+    #[arg(long)]
+    from_peer: String,
+    /// Upstream peer's RPC socket.
+    #[arg(long)]
+    from_addr: SocketAddr,
+    /// Downstream peer name.
+    #[arg(long)]
+    to_peer: String,
+    /// Downstream peer's RPC socket.
+    #[arg(long)]
+    to_addr: SocketAddr,
+    /// mTLS material for BOTH connections. Assumes upstream +
+    /// downstream share the same fleet CA (the common case) — a
+    /// mixed-fleet variant with separate identities would need a
+    /// second `--tls-dir` and lives in a future extension.
+    #[arg(long)]
+    tls_dir: PathBuf,
+    /// Tag to copy (e.g. `clawverse:main:latest`). Required for now;
+    /// fingerprint-based copies land later once the caller knows what
+    /// fingerprint to prewarm.
+    #[arg(long)]
+    pin: String,
 }
 
 #[derive(clap::Args, Debug, Clone)]
@@ -184,6 +217,7 @@ async fn main() -> Result<()> {
         Cmd::Pin(args) => cmd_pin(args).await,
         Cmd::Unpin(args) => cmd_unpin(args).await,
         Cmd::ListTags(args) => cmd_list_tags(args).await,
+        Cmd::Prewarm(args) => cmd_prewarm(args).await,
     }
 }
 
@@ -650,6 +684,121 @@ async fn _reserved_call_get_tag(conn: &quinn::Connection, key: &str) -> Result<O
     call_get_tag(conn, key).await
 }
 
+// ── prewarm ──────────────────────────────────────────────────────────
+
+async fn cmd_prewarm(args: PrewarmArgs) -> Result<()> {
+    let identity = NodeIdentity::from_pem_dir(&args.tls_dir).with_context(|| {
+        format!("loading node identity from {}", args.tls_dir.display())
+    })?;
+
+    // Two clients — one per direction. Both share the same identity
+    // (single fleet CA); a mixed-fleet variant would need a second
+    // `--tls-dir` for the downstream side.
+    let up_client = QuicClient::new("0.0.0.0:0".parse()?, identity)?;
+    let up_conn = up_client
+        .connect(args.from_addr, &args.from_peer)
+        .await
+        .with_context(|| format!("connecting to upstream {} @ {}", args.from_peer, args.from_addr))?;
+
+    // Resolve the tag upstream → BlobId → BlobStat.
+    let (blob_id, stat) = match resolve_pin(&up_conn, &args.pin).await? {
+        Some(pair) => pair,
+        None => {
+            up_conn.close(quinn::VarInt::from_u32(0), b"done");
+            up_client.shutdown().await;
+            anyhow::bail!(
+                "upstream {} has no tag {:?}",
+                args.from_peer,
+                args.pin
+            );
+        }
+    };
+    tracing::info!(
+        "prewarm: upstream tag {} → blob {} ({} bytes, {} chunks)",
+        args.pin,
+        blob_id,
+        stat.total_size,
+        stat.chunk_count
+    );
+
+    // Download the whole blob. Buffered — the streaming zero-copy
+    // upstream→downstream pipe is a follow-on that avoids the RAM
+    // ceiling (~16 MiB currently) but complicates task ownership.
+    let download_started = Instant::now();
+    let mut buf: Vec<u8> = Vec::with_capacity(stat.total_size as usize);
+    let ok = call_blob_get_stream(&up_conn, &blob_id, &mut buf).await?;
+    if !ok {
+        up_conn.close(quinn::VarInt::from_u32(0), b"done");
+        up_client.shutdown().await;
+        anyhow::bail!(
+            "upstream reported tag {} → blob {} but BlobGetStream returned NotFound",
+            args.pin,
+            blob_id
+        );
+    }
+    let download_elapsed = download_started.elapsed();
+    tracing::info!(
+        "prewarm: downloaded {} bytes from upstream in {:?}",
+        buf.len(),
+        download_elapsed
+    );
+
+    // Second client for the downstream side. Distinct QUIC endpoint
+    // so we don't conflate connection state.
+    let down_identity = NodeIdentity::from_pem_dir(&args.tls_dir)?;
+    let down_client = QuicClient::new("0.0.0.0:0".parse()?, down_identity)?;
+    let down_conn = down_client
+        .connect(args.to_addr, &args.to_peer)
+        .await
+        .with_context(|| format!("connecting to downstream {} @ {}", args.to_peer, args.to_addr))?;
+
+    let upload_started = Instant::now();
+    let cursor = std::io::Cursor::new(buf.clone());
+    let assigned_id = call_blob_put_stream(&down_conn, cursor).await?;
+    if assigned_id != blob_id {
+        down_conn.close(quinn::VarInt::from_u32(0), b"done");
+        up_conn.close(quinn::VarInt::from_u32(0), b"done");
+        down_client.shutdown().await;
+        up_client.shutdown().await;
+        anyhow::bail!(
+            "prewarm integrity check failed: downstream stored {} but upstream had {}",
+            assigned_id,
+            blob_id
+        );
+    }
+    let upload_elapsed = upload_started.elapsed();
+    tracing::info!(
+        "prewarm: uploaded {} bytes to downstream in {:?}",
+        buf.len(),
+        upload_elapsed
+    );
+
+    // Publish the tag downstream too — otherwise a `prefetch --pin`
+    // there would still miss.
+    call_put_tag(&down_conn, &args.pin, blob_id.as_bytes()).await?;
+
+    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;
+
+    println!();
+    println!("── claw-cargo prewarm ──────────────────────────────");
+    println!("tag:         {}", args.pin);
+    println!("blob:        {}", blob_id);
+    println!("upstream:    {} @ {}", args.from_peer, args.from_addr);
+    println!("downstream:  {} @ {}", args.to_peer, args.to_addr);
+    println!(
+        "bytes:       {} ({} chunks)",
+        stat.total_size, stat.chunk_count
+    );
+    println!("download:    {:?}", download_elapsed);
+    println!("upload:      {:?}", upload_elapsed);
+    println!("total:       {:?}", download_elapsed + upload_elapsed);
+    println!("────────────────────────────────────────────────────");
+    Ok(())
+}
+
 fn run_cargo(
     workspace: &Path,
     profile: &str,

claw-store/src/cluster/rpc.rs

@@ -771,3 +771,7 @@ pub use client::*;
 #[cfg(test)]
 #[path = "rpc/tests.rs"]
 mod tests;
+
+#[cfg(test)]
+#[path = "rpc/tests_phase5.rs"]
+mod tests_phase5;

claw-store/src/cluster/rpc/tests.rs

@@ -725,579 +725,3 @@ async fn end_to_end_push_blob_missing_chunks_replicates_only_needed_bytes() {
     accept_task.abort();
 }
 
-// ── 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]);
-}
-
-#[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();
-}

claw-store/src/cluster/rpc/tests_phase5.rs

@@ -0,0 +1,722 @@
+//! 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]);
+}
+
+#[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();
+}