claw-store/src/cluster/refs.rs

@@ -31,6 +31,85 @@ pub type RefKey = [u8; 32];
 /// A raw 32-byte value.
 pub type RefValue = [u8; 32];
 
+/// Phase 3 (2026-07-13): 8-byte node identifier used as a tie-breaker
+/// in the CRDT merge order. Convention: `blake3(node_name)[0..8]`.
+/// Two nodes will only collide on this if they share the same 8-byte
+/// prefix of their names' hashes (birthday-bound at ~4B).
+pub type NodeStamp = [u8; 8];
+
+/// A ref value paired with its Lamport clock and originator node
+/// stamp — the tuple that Phase 3 CRDT semantics needs to merge
+/// concurrent writes deterministically.
+///
+/// The pair `(clock, node)` forms a total order — `dominates`
+/// returns true when `self` is strictly newer than `other`. Equal
+/// pairs are treated as "same write, idempotent" (no-op).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct StampedRef {
+    pub value: RefValue,
+    pub clock: u64,
+    pub node: NodeStamp,
+}
+
+impl StampedRef {
+    /// 48-byte on-disk / on-wire representation: `value:32 || clock:u64 LE || node:8`.
+    pub const ENCODED_LEN: usize = 48;
+
+    pub fn to_bytes(&self) -> [u8; Self::ENCODED_LEN] {
+        let mut out = [0u8; Self::ENCODED_LEN];
+        out[..32].copy_from_slice(&self.value);
+        out[32..40].copy_from_slice(&self.clock.to_le_bytes());
+        out[40..48].copy_from_slice(&self.node);
+        out
+    }
+
+    pub fn from_bytes(bytes: &[u8]) -> Result<Self> {
+        if bytes.len() != Self::ENCODED_LEN {
+            bail!(
+                "stamped ref wrong length {} (expected {})",
+                bytes.len(),
+                Self::ENCODED_LEN
+            );
+        }
+        let mut value = [0u8; 32];
+        value.copy_from_slice(&bytes[..32]);
+        let mut clock_bytes = [0u8; 8];
+        clock_bytes.copy_from_slice(&bytes[32..40]);
+        let mut node = [0u8; 8];
+        node.copy_from_slice(&bytes[40..48]);
+        Ok(Self {
+            value,
+            clock: u64::from_le_bytes(clock_bytes),
+            node,
+        })
+    }
+
+    /// Strict CRDT merge order: `self` beats `other` iff the pair
+    /// `(clock, node)` is strictly greater.
+    pub fn dominates(&self, other: &Self) -> bool {
+        (self.clock, self.node) > (other.clock, other.node)
+    }
+}
+
+/// Derive a stable 8-byte node stamp from a human node name.
+pub fn node_stamp_for(name: &str) -> NodeStamp {
+    let hash = blake3::hash(name.as_bytes());
+    let mut out = [0u8; 8];
+    out.copy_from_slice(&hash.as_bytes()[..8]);
+    out
+}
+
+/// Outcome of a stamped put. Callers can distinguish "we accepted
+/// your write" from "someone else already had a newer/equal write".
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum PutOutcome {
+    /// Incoming write was strictly newer; on-disk value updated.
+    Merged,
+    /// A prior write with `(clock, node) >= incoming` already exists.
+    /// The on-disk value is unchanged; `current` is what's there.
+    Rejected { current: StampedRef },
+}
+
 /// Directory-backed reference store.
 #[derive(Debug, Clone)]
 pub struct RefStore {
@@ -117,6 +196,61 @@ impl RefStore {
             .join(format!("{hex}.ref"))
     }
 
+    /// Phase 3 (2026-07-13): Lamport-stamped ref lookup. Reads from
+    /// the `refs-v2/` directory (separate namespace from the raw
+    /// `refs/` set) so the two lookup surfaces don't interfere.
+    /// Returns `None` when no stamped ref exists.
+    pub async fn get_stamped(&self, key: &RefKey) -> Result<Option<StampedRef>> {
+        let path = self.stamped_path(key);
+        match tokio::fs::read(&path).await {
+            Ok(bytes) => Ok(Some(StampedRef::from_bytes(&bytes).with_context(|| {
+                format!("decoding stamped ref at {}", path.display())
+            })?)),
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
+            Err(e) => Err(anyhow::Error::from(e))
+                .with_context(|| format!("reading stamped ref at {}", path.display())),
+        }
+    }
+
+    /// Phase 3 (2026-07-13): merge an incoming stamped ref against
+    /// the local view.
+    ///
+    /// Merge rule: the pair `(clock, node)` forms a total order.
+    /// The higher pair wins. Equal pairs are idempotent — the write
+    /// is treated as a no-op and reports [`PutOutcome::Rejected`]
+    /// with the current value (so callers can distinguish "already
+    /// have it" from "someone raced us").
+    ///
+    /// Never lowers the on-disk value: a stale write from a partitioned
+    /// peer is simply ignored.
+    pub async fn put_stamped(
+        &self,
+        key: &RefKey,
+        incoming: StampedRef,
+    ) -> Result<PutOutcome> {
+        let path = self.stamped_path(key);
+        if let Some(current) = self.get_stamped(key).await? {
+            if !incoming.dominates(&current) {
+                return Ok(PutOutcome::Rejected { current });
+            }
+        }
+        if let Some(parent) = path.parent() {
+            tokio::fs::create_dir_all(parent).await.with_context(|| {
+                format!("creating stamped-ref bucket {}", parent.display())
+            })?;
+        }
+        self.atomic_write(&path, &incoming.to_bytes()).await?;
+        Ok(PutOutcome::Merged)
+    }
+
+    fn stamped_path(&self, key: &RefKey) -> PathBuf {
+        let hex = hex32(key);
+        self.root
+            .join("refs-v2")
+            .join(&hex[..2])
+            .join(format!("{hex}.svref"))
+    }
+
     async fn atomic_write(&self, final_path: &Path, bytes: &[u8]) -> Result<()> {
         let tmp_dir = self.root.join(".tmp");
         let tmp_name = format!(
@@ -240,4 +374,137 @@ mod tests {
         let err = store.get(&key).await.unwrap_err().to_string();
         assert!(err.contains("wrong length"), "unexpected: {err}");
     }
+
+    // ── Phase 3: stamped refs ────────────────────────────────────────
+
+    fn stamped(value: u8, clock: u64, node: u8) -> StampedRef {
+        StampedRef {
+            value: [value; 32],
+            clock,
+            node: [node; 8],
+        }
+    }
+
+    #[test]
+    fn stamped_ref_encoding_round_trips() {
+        let r = stamped(7, 42, 3);
+        let bytes = r.to_bytes();
+        assert_eq!(bytes.len(), StampedRef::ENCODED_LEN);
+        let back = StampedRef::from_bytes(&bytes).unwrap();
+        assert_eq!(back, r);
+    }
+
+    #[test]
+    fn stamped_ref_from_bytes_rejects_wrong_length() {
+        let err = StampedRef::from_bytes(&[0u8; 32])
+            .unwrap_err()
+            .to_string();
+        assert!(err.contains("wrong length"), "unexpected: {err}");
+    }
+
+    #[test]
+    fn dominates_is_a_total_order_on_clock_then_node() {
+        // Higher clock always dominates.
+        let a = stamped(1, 5, 1);
+        let b = stamped(2, 6, 0);
+        assert!(b.dominates(&a));
+        assert!(!a.dominates(&b));
+
+        // Same clock: higher node stamp wins.
+        let c = stamped(3, 10, 1);
+        let d = stamped(4, 10, 2);
+        assert!(d.dominates(&c));
+        assert!(!c.dominates(&d));
+
+        // Same everything: neither dominates (idempotent).
+        assert!(!c.dominates(&c));
+    }
+
+    #[test]
+    fn node_stamp_is_stable_for_same_name() {
+        let a = node_stamp_for("tank");
+        let b = node_stamp_for("tank");
+        assert_eq!(a, b);
+        // Different name → different stamp (birthday-bound; tank vs architect definitely differs).
+        assert_ne!(a, node_stamp_for("architect"));
+    }
+
+    #[tokio::test]
+    async fn put_stamped_and_get_stamped_round_trip() {
+        let (_tmp, store) = open();
+        let key = [9u8; 32];
+        assert_eq!(store.get_stamped(&key).await.unwrap(), None);
+        let s = stamped(1, 1, 1);
+        assert!(matches!(
+            store.put_stamped(&key, s).await.unwrap(),
+            PutOutcome::Merged
+        ));
+        assert_eq!(store.get_stamped(&key).await.unwrap(), Some(s));
+    }
+
+    #[tokio::test]
+    async fn put_stamped_rejects_older_and_equal_writes() {
+        // Older by clock: rejected. Equal (clock, node): rejected
+        // (idempotent). Newer by clock or by node-stamp: merged.
+        let (_tmp, store) = open();
+        let key = [0xAB; 32];
+        let base = stamped(1, 10, 5);
+        store.put_stamped(&key, base).await.unwrap();
+
+        // Older clock → rejected, current returned.
+        let older = stamped(2, 9, 9);
+        match store.put_stamped(&key, older).await.unwrap() {
+            PutOutcome::Rejected { current } => assert_eq!(current, base),
+            _ => panic!("expected Rejected"),
+        }
+        assert_eq!(store.get_stamped(&key).await.unwrap(), Some(base));
+
+        // Equal (clock, node) with different value → rejected. Prevents
+        // "value drift" from a concurrent writer at the same tick.
+        let dupe = StampedRef {
+            value: [0xEE; 32],
+            clock: base.clock,
+            node: base.node,
+        };
+        match store.put_stamped(&key, dupe).await.unwrap() {
+            PutOutcome::Rejected { current } => assert_eq!(current.value, base.value),
+            _ => panic!("expected Rejected on equal (clock,node)"),
+        }
+
+        // Same clock but higher node stamp → merged.
+        let same_clock_higher_node = StampedRef {
+            value: [0x11; 32],
+            clock: base.clock,
+            node: [0x99; 8],
+        };
+        assert!(matches!(
+            store.put_stamped(&key, same_clock_higher_node).await.unwrap(),
+            PutOutcome::Merged
+        ));
+        assert_eq!(
+            store.get_stamped(&key).await.unwrap(),
+            Some(same_clock_higher_node)
+        );
+
+        // Higher clock always wins regardless of node.
+        let higher_clock = stamped(0x22, base.clock + 1, 0);
+        assert!(matches!(
+            store.put_stamped(&key, higher_clock).await.unwrap(),
+            PutOutcome::Merged
+        ));
+        assert_eq!(store.get_stamped(&key).await.unwrap(), Some(higher_clock));
+    }
+
+    #[tokio::test]
+    async fn stamped_and_unstamped_stores_are_independent() {
+        // put() writes to refs/, put_stamped() writes to refs-v2/.
+        // They must not clobber each other.
+        let (_tmp, store) = open();
+        let key = [0xCC; 32];
+        store.put(&key, &[0xAA; 32]).await.unwrap();
+        let s = stamped(0xBB, 1, 1);
+        store.put_stamped(&key, s).await.unwrap();
+        assert_eq!(store.get(&key).await.unwrap(), Some([0xAA; 32]));
+        assert_eq!(store.get_stamped(&key).await.unwrap(), Some(s));
+    }
 }

claw-store/src/cluster/rpc.rs

@@ -28,7 +28,7 @@
 use crate::cluster::blob::{BlobId, BlobManifest, BlobStat, BlobStore, ChunkHash};
 use crate::cluster::gossip::{ClusterGossip, PeerView};
 use crate::cluster::metrics::{CacheMetrics, MetricsReply};
-use crate::cluster::refs::{RefKey, RefStore, RefValue};
+use crate::cluster::refs::{PutOutcome, RefKey, RefStore, RefValue, StampedRef};
 use crate::cluster::tags::{TagEntry, TagStore};
 use anyhow::{bail, Context, Result};
 use quinn::{Connection, ConnectionError};
@@ -137,6 +137,22 @@ pub enum Method {
     /// `payload`: 32-byte `RefKey`. Reply: 32 bytes on hit or
     /// single-byte [`ErrorCode::NotFound`].
     GetRefLocal = 0x14,
+    /// Phase 3 (2026-07-13): stamped/versioned PutRef. Merges under a
+    /// Lamport-clock + node-stamp total order — concurrent writers
+    /// can no longer clobber each other silently.
+    /// `payload`: 32-byte `RefKey` || 32-byte `RefValue` ||
+    /// 8-byte little-endian `u64` clock || 8-byte node stamp
+    /// (`blake3(node_name)[0..8]`).
+    /// Reply: single-byte status —
+    ///   `STREAM_STATUS_OK` = accepted (Merged);
+    ///   [`ErrorCode::AlreadyExists`] = rejected (older/equal).
+    PutRefVersioned = 0x15,
+    /// Phase 3: stamped GetRef. Returns the current
+    /// (value, clock, node) triple as 48 bytes.
+    /// `payload`: 32-byte `RefKey`.
+    /// Reply: 48 bytes on hit, single-byte
+    /// [`ErrorCode::NotFound`] on miss.
+    GetRefVersioned = 0x16,
 }
 
 impl Method {
@@ -164,6 +180,8 @@ impl Method {
             0x12 => Some(Method::ListTags),
             0x13 => Some(Method::GetMetrics),
             0x14 => Some(Method::GetRefLocal),
+            0x15 => Some(Method::PutRefVersioned),
+            0x16 => Some(Method::GetRefVersioned),
             _ => None,
         }
     }
@@ -199,6 +217,11 @@ pub enum ErrorCode {
     /// configured (e.g. Blob RPCs called on a node with no local
     /// blob store).
     NotConfigured = 0xf5,
+    /// Phase 3 (2026-07-13): the write was rejected because a prior
+    /// value with an equal or higher `(clock, node)` already exists.
+    /// The current value stayed on disk. Used by
+    /// [`Method::PutRefVersioned`] to signal a "not merged" outcome.
+    AlreadyExists = 0xf6,
 }
 
 impl ErrorCode {
@@ -213,6 +236,9 @@ impl ErrorCode {
             ErrorCode::NotFound => "not found",
             ErrorCode::InvalidRequest => "invalid request",
             ErrorCode::NotConfigured => "server subsystem not configured",
+            ErrorCode::AlreadyExists => {
+                "rejected: existing value dominates incoming write"
+            }
         }
     }
 }
@@ -609,6 +635,42 @@ impl RpcRouter {
                 store.put(&key, &value).await?;
                 Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK]))
             }
+            Method::PutRefVersioned => {
+                let store = match &self.ref_store {
+                    Some(s) => s,
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
+                };
+                // Wire: 32-key || 32-value || 8-clock (LE) || 8-node = 80.
+                if payload.len() != 32 + StampedRef::ENCODED_LEN {
+                    return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest));
+                }
+                let mut key = [0u8; 32];
+                key.copy_from_slice(&payload[..32]);
+                let incoming = match StampedRef::from_bytes(&payload[32..]) {
+                    Ok(v) => v,
+                    Err(_) => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
+                };
+                match store.put_stamped(&key, incoming).await? {
+                    PutOutcome::Merged => Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK])),
+                    PutOutcome::Rejected { .. } => {
+                        Ok(HandlerOutcome::Error(ErrorCode::AlreadyExists))
+                    }
+                }
+            }
+            Method::GetRefVersioned => {
+                let store = match &self.ref_store {
+                    Some(s) => s,
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
+                };
+                let key = match decode_32(payload) {
+                    Some(k) => k,
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
+                };
+                match store.get_stamped(&key).await? {
+                    Some(s) => Ok(HandlerOutcome::Reply(s.to_bytes().to_vec())),
+                    None => Ok(HandlerOutcome::Error(ErrorCode::NotFound)),
+                }
+            }
             Method::PutTag => {
                 let store = match &self.tag_store {
                     Some(s) => s,

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();
+}