Phase 5g: cache metrics + GetMetrics RPC + peer-metrics CLI

Every RPC handler that answers a hit-or-miss question now increments
lock-free atomic counters. The GetMetrics RPC (0x13) returns a JSON
snapshot of every counter; the new claw-cargo peer-metrics CLI
prints hit rates, byte volumes, and counter uptime.

Placement engines can now poll these across the fleet to bias runner
scheduling toward whichever node has the warmest cache for a given
repo/tag combination.

## New module: cluster/metrics.rs (268 lines)

Types:
- CacheMetrics — atomic counters, all AtomicU64, Relaxed ordering
  (metrics are advisory, not consistency-critical)
- MetricsReply — JSON snapshot returned by GetMetrics

Public API:
- CacheMetrics::new() — timestamped start, all counters at 0
- record_get_ref_hit / _miss
- record_get_tag_hit / _miss
- record_blob_get_bytes / record_blob_put_bytes
- record_get_chunk_hit / _miss
- record_has_chunk_hit / _miss
- snapshot() — atomic-load every field into a MetricsReply

MetricsReply derived helpers:
- get_ref_hit_rate() / get_tag_hit_rate() / has_chunk_hit_rate() —
  Option<f64> so 0/0 returns None instead of NaN

## RPC method

- GetMetrics (0x13): payload = empty; reply = JSON MetricsReply

Wire-level instrumentation added to RpcRouter dispatch:
- GetRef  → record_get_ref_hit / _miss
- GetTag  → record_get_tag_hit / _miss
- BlobGet → record_blob_get_bytes (on hit)
- BlobPut → record_blob_put_bytes
- HasChunk → record_has_chunk_hit / _miss
- GetChunk → record_get_chunk_hit + record_blob_get_bytes on hit
             / record_get_chunk_miss

RpcRouter grows Arc<CacheMetrics> unconditionally — every router has
metrics, so a fresh node with no traffic still returns a valid
snapshot with all zeros + started_unix.

Streaming variants (BlobPutStream / BlobGetStream) don't yet track
byte counts — they'd require plumbing the count out of put_stream /
stream_to. Follow-on if it turns out to matter for placement.

## Client helper + CLI

- call_get_metrics(&conn) → Result<MetricsReply>

- claw-cargo peer-metrics [--peer ...] [--peer-addr ...] [--tls-dir ...]
    Fetches + pretty-prints:
      counter uptime:      42s
      GetRef  hits/misses: 123 / 45
              hit rate:    73.21%
      GetTag  hits/misses: 8 / 2
              hit rate:    80.00%
      HasChunk hits/miss:  512 / 88
               hit rate:   85.33%
      GetChunk hits/miss:  47 / 12
      Blob GET bytes:      1.23 GiB
      Blob PUT bytes:      3.45 GiB

human_bytes() helper picks GiB / MiB / KiB / B based on magnitude.

Subcommand count now 9: build / prefetch / status / fingerprint /
pin / unpin / list-tags / prewarm / peer-metrics.

## Tests (13 new, all real filesystem / real QUIC — no mocks)

CacheMetrics (6):
- new_starts_all_counters_at_zero_except_timestamp
- recorders_increment_the_right_field (every recorder × 1-2 counts)
- hit_rates_none_when_zero_events (avoids 0/0 NaN)
- hit_rates_compute_correctly (3 hits / 1 miss → 75%)
- snapshot_round_trips_through_json
- snapshots_across_threads_are_consistent_up_to_relaxed_ordering
  (8 threads × 1000 increments → exactly 8000)

RPC integration (7):
- phase_5g_method_byte_encoding
- get_metrics_returns_empty_snapshot_before_any_activity
- get_ref_records_hit_and_miss_counters (2 hits + 1 miss)
- get_tag_records_hit_and_miss_counters
- blob_get_and_blob_put_record_byte_counts
- has_chunk_and_get_chunk_record_hit_miss_counters
- **end_to_end_get_metrics_over_real_quic** — seed activity locally,
  fire GetRef/GetTag/BlobGet dispatches to move the counters, then
  fetch metrics through real QUIC + mTLS and verify each field
  including the 0.5 hit rate calculation

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

File sizes (all under 1300-line ceiling):
- cluster/metrics.rs: 268
- cluster/rpc.rs: 818
- cluster/rpc/tests_phase5.rs: 905
- claw_cargo.rs: 894

## What this enables

Fleet-wide visibility into which peer is actually serving traffic:

  # From anywhere with connectivity + fleet mTLS
  claw-cargo peer-metrics --peer tank
  claw-cargo peer-metrics --peer architect
  claw-cargo peer-metrics --peer morpheus

Compare hit rates side by side to see which node's cache is warmest.
A placement engine can automate this — poll every 30s, feed the
scheduler.

## Follow-on

- 5h: streaming variant of prewarm (fixed-memory ceiling for many-GB
  blobs)
- 5i: metrics also published via gossip so PeerView carries hit rate
  without a per-peer GetMetrics roundtrip
- 5j: prometheus /metrics endpoint on the daemon for existing dash
  integrations
- 6: FUSE mount for warm-tier git worktrees

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

@@ -290,6 +290,189 @@ async fn list_tags_returns_json_sorted() {
     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`.