Phase 5b: KV refs + claw-cargo CLI (the killer feature, live)

Ships the actual user-facing cargo build cache. Combined with Phase 5a
(fingerprint + capture + restore) + the whole Phase 2 blob substrate,
`claw-cargo build` now runs `cargo build` with a peer-cache lookup:
hit → download+restore, miss → build+capture+upload.

## What ships

### cluster/refs.rs (243 lines)

A dumb 32-byte-key → 32-byte-value directory-backed store. Used to map
fingerprints → BlobIds. Layout mirrors BlobStore:

  <root>/
    refs/<kk>/<key_hex>.ref     — 32 raw bytes
    .tmp/                        — atomic-rename staging

Public API: RefStore::open / get / put / delete / contains. All writes
atomic via tempfile + rename. Deliberately no versioning or CRDT
semantics — that's Phase 3. Every real cargo-cache lookup is a
single-key-single-value shape.

### New RPC methods

- GetRef (0x0d): payload = 32-byte RefKey; reply = 32 bytes / NotFound
- PutRef (0x0e): payload = 32-byte RefKey || 32-byte RefValue;
  reply = STREAM_STATUS_OK / error

### RpcRouter + services

- RpcRouter grows optional Arc<RefStore> via `with_ref_store`
- ClusterServices opens a RefStore alongside the BlobStore when
  `blob_store_root` is configured (co-located at `<blob_root>/refs-db`)
- `blob_store_enabled()` / `ref_store_enabled()` introspection

### claw-cargo binary (319 lines)

New bin target `claw-cargo` — thin CLI wrapping the whole stack:

  claw-cargo fingerprint --profile release --features "a,b"
    → prints the workspace fingerprint (no network)

  claw-cargo build \
    --peer <name> --peer-addr <ip:port> --tls-dir <dir> \
    --profile release --features "a,b" \
    -- --workspace=x --frozen ...
    → 1. compute fingerprint
      2. QUIC + mTLS connect to peer
      3. GetRef(fingerprint) → BlobId?
         HIT: BlobStat → BlobGetStream → restore_target → cargo build
         MISS: cargo build → capture_target → BlobPutStream → PutRef
      4. Print summary: fingerprint, hit/miss, bytes, cargo elapsed

## Live smoke test

Ran claw-cargo fingerprint on this workspace with three profile/feature
combos — got three distinct 32-byte fingerprints. Same profile+features
on the same workspace state → same fingerprint (Phase 5a's guarantee
carried through the CLI).

## Tests (14 new, all real — no mocks)

Refs store (7):
- open creates layout
- get returns None for missing
- put + get round-trips
- put overwrites prior value
- delete removes ref + reports (false on second delete)
- distinct keys produce distinct on-disk files (bucket fan-out proof)
- rejects_wrong_length_on_disk (corruption detection)

RPC (7):
- phase_5b_method_byte_encoding
- get_ref_returns_not_found_for_missing
- put_ref_stores_and_get_ref_reads_back
- put_ref_rejects_wrong_length_payload
- get_ref_rejects_wrong_length_payload
- ref_rpcs_return_not_configured_without_store
- end_to_end_put_ref_get_ref_over_real_quic — full 2-node QUIC + mTLS
  round trip proving PutRef/GetRef work at the wire level

188 tests pass. Pre-existing macOS-only failure unchanged.

File sizes (all under 1300-line ceiling):
- cluster/refs.rs: 243
- cluster/rpc.rs: 1169
- cluster/rpc/tests.rs: 1073
- cluster/services.rs: 565
- claw_cargo.rs: 319

## Where this leaves us

The distributed FS + cargo cache is functionally complete for the
happy path:

  Node A builds clawverse for the first time
  → cargo build (50 min cold)
  → capture_target (a few seconds)
  → push to node B via BlobPutStream (network-bound)
  → PutRef(fingerprint → BlobId)

  Node B on the same workspace state runs `claw-cargo build …`
  → compute_fingerprint (ms)
  → GetRef → hit
  → BlobGetStream (network-bound)
  → restore_target (a few seconds)
  → cargo build → sees valid deps/.fingerprint, builds only
    workspace crates (~3 min instead of 50)

Same workspace state on a third machine? Same fingerprint → same
cache hit. That's the whole design.

## Follow-on

- Phase 5c: pre-fetch on Gitea webhook so CI runners never wait
- Phase 5d: metric ticker publishes cache hit rate into gossip so
  the placement engine can bias runner scheduling toward warm nodes
- Phase 3: CRDT metadata for human-readable pins on top of raw
  32-byte refs (`clawverse:main:latest-cache` → fingerprint hex)
- Phase 6+: FUSE mount for the warm-tier git worktrees

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

@@ -725,6 +725,139 @@ 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();
+}
+
 #[tokio::test]
 async fn call_get_chunk_verifies_returned_hash() {
     let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();