Phase 3a: Lamport-stamped refs with CRDT-merge on PutRef

Concurrent PutRef safety via (clock, node) total order. New wire
methods PutRefVersioned (0x15) + GetRefVersioned (0x16). Existing
PutRef/GetRef unchanged for backward compat. Data in refs-v2/
namespace so the two coexist during cutover.

+8 tests, 272 total (baseline +8).

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

@@ -180,6 +180,7 @@ pub fn decode_error(b: u8) -> Option<ErrorCode> {
         0xf3 => Some(ErrorCode::NotFound),
         0xf4 => Some(ErrorCode::InvalidRequest),
         0xf5 => Some(ErrorCode::NotConfigured),
+        0xf6 => Some(ErrorCode::AlreadyExists),
         _ => None,
     }
 }
@@ -472,6 +473,61 @@ pub async fn call_put_ref(
     }
 }
 
+// ── Phase 3: stamped-ref client helpers ──────────────────────────────
+
+/// Phase 3 (2026-07-13): submit a stamped (CRDT-merge) PutRef.
+///
+/// Returns `Ok(true)` when the peer merged the write (Merged),
+/// `Ok(false)` when the peer rejected it because an equal or
+/// higher `(clock, node)` already exists (AlreadyExists). Any
+/// other reply is an error.
+pub async fn call_put_ref_versioned(
+    conn: &Connection,
+    key: &crate::cluster::refs::RefKey,
+    incoming: &crate::cluster::refs::StampedRef,
+) -> Result<bool> {
+    let mut payload = Vec::with_capacity(32 + crate::cluster::refs::StampedRef::ENCODED_LEN);
+    payload.extend_from_slice(key);
+    payload.extend_from_slice(&incoming.to_bytes());
+    let reply = rpc_call(conn, Method::PutRefVersioned, &payload).await?;
+    if reply.len() != 1 {
+        bail!(
+            "expected single-byte PutRefVersioned reply, got {} bytes",
+            reply.len()
+        );
+    }
+    match reply[0] {
+        STREAM_STATUS_OK => Ok(true),
+        code => match decode_error(code) {
+            Some(ErrorCode::AlreadyExists) => Ok(false),
+            Some(err) => bail!("peer rejected PutRefVersioned: {}", err.describe()),
+            None => bail!(
+                "peer replied with unknown byte 0x{:02x} for PutRefVersioned",
+                code
+            ),
+        },
+    }
+}
+
+/// Phase 3: fetch a stamped (CRDT-merge) ref.
+pub async fn call_get_ref_versioned(
+    conn: &Connection,
+    key: &crate::cluster::refs::RefKey,
+) -> Result<Option<crate::cluster::refs::StampedRef>> {
+    let reply = rpc_call(conn, Method::GetRefVersioned, key).await?;
+    if reply.len() == 1 {
+        match decode_error(reply[0]) {
+            Some(ErrorCode::NotFound) => return Ok(None),
+            Some(err) => bail!("peer replied with error: {}", err.describe()),
+            None => {}
+        }
+    }
+    Ok(Some(
+        crate::cluster::refs::StampedRef::from_bytes(&reply)
+            .context("decoding stamped ref reply")?,
+    ))
+}
+
 // ── Phase 5g: cache metrics client helper ────────────────────────────
 
 /// Fetch the peer's current cache-metrics snapshot.

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

@@ -1273,3 +1273,94 @@ async fn parallel_blob_get_reassembles_multi_chunk_blob_byte_equal() {
     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();
+}