claw-store/src/claw_cargo.rs

@@ -47,8 +47,8 @@ use crate::cluster::build_cache::{
 use crate::cluster::client_config::{ClientConfig, ResolvedClientConfig};
 use crate::cluster::rpc::{
     call_blob_get_parallel, call_blob_get_stream, call_blob_put_stream, call_blob_stat,
-    call_delete_tag, call_get_metrics, call_peer_status,
+    call_delete_tag, call_get_metrics, call_get_ref_versioned, call_peer_status,
-    prewarm_missing_chunks_between_parallel,
+    call_put_ref_versioned, prewarm_missing_chunks_between_parallel,
     call_get_ref, call_get_tag, call_list_tags, call_put_ref, call_put_tag,
 };
 use crate::cluster::transport::{NodeIdentity, QuicClient};
@@ -324,16 +324,21 @@ async fn peer_lookup(
     fp: &Fingerprint,
 ) -> Result<Option<(BlobId, crate::cluster::blob::BlobStat)>> {
     let key = *fp.as_bytes();
-    let value_bytes = match call_get_ref(conn, &key).await? {
+    // Phase 3b (2026-07-13): read stamped refs first so cross-runner
-        Some(v) => v,
+    // ref-forwarding uses the CRDT-merge path. Fall back to legacy
-        None => return Ok(None),
+    // unstamped refs for any pre-Phase-3a data still on disk.
-    };
+    let value_bytes: crate::cluster::refs::RefValue =
+        match call_get_ref_versioned(conn, &key).await? {
+            Some(s) => s.value,
+            None => match call_get_ref(conn, &key).await? {
+                Some(v) => v,
+                None => return Ok(None),
+            },
+        };
     let blob_id = BlobId::from_bytes(value_bytes);
     match call_blob_stat(conn, &blob_id).await? {
         Some(stat) => Ok(Some((blob_id, stat))),
         None => {
-            // Ref points at a missing blob — treat as miss so
-            // build+upload will re-populate.
             tracing::warn!(
                 "ref pointed at blob {} but peer has no such blob; treating as miss",
                 blob_id
@@ -664,8 +669,27 @@ async fn cmd_build(args: BuildArgs) -> Result<()> {
                 .await
                 .context("re-opening capture tempfile for upload")?;
             let blob_id = call_blob_put_stream(&conn, reader).await?;
-            call_put_ref(&conn, fp.as_bytes(), blob_id.as_bytes()).await?;
+            // Phase 3b: stamped write — concurrent PutRef races are
-            tracing::info!("uploaded blob {} + set ref", blob_id);
+            // resolved by (unix_secs, blake3(hostname)[..8]).
+            // AlreadyExists is fine: the winner beat us to it and
+            // its blob is byte-identical (content-addressed).
+            let stamped = build_stamped_ref(&blob_id);
+            let merged =
+                call_put_ref_versioned(&conn, fp.as_bytes(), &stamped).await?;
+            if merged {
+                tracing::info!(
+                    "uploaded blob {} + set stamped ref (clock={})",
+                    blob_id,
+                    stamped.clock
+                );
+            } else {
+                tracing::info!(
+                    "uploaded blob {} but a concurrent writer already \
+                     published a dominant ref — ok, content-addressed \
+                     blob is identical",
+                    blob_id
+                );
+            }
             outcome = CacheOutcome::Populated {
                 blob_id,
                 uploaded_bytes: capture_bytes,
@@ -1046,7 +1070,14 @@ async fn cmd_prewarm(args: PrewarmArgs) -> Result<()> {
     let fingerprint_published = match call_get_tag(&up_conn, &companion).await? {
         Some(fp_bytes) => {
             call_put_tag(&down_conn, &companion, &fp_bytes).await?;
-            call_put_ref(&down_conn, &fp_bytes, blob_id.as_bytes()).await?;
+            // Phase 3b: stamped ref on the downstream. The tag itself
+            // is unstamped (single-owner semantics), but the
+            // fingerprint→blob ref is CRDT-merged so a concurrent
+            // runner on the downstream doesn't clobber a prewarm.
+            let mut key = [0u8; 32];
+            key.copy_from_slice(&fp_bytes);
+            let stamped = build_stamped_ref(&blob_id);
+            let _ = call_put_ref_versioned(&down_conn, &key, &stamped).await?;
             true
         }
         None => false,
@@ -1122,6 +1153,53 @@ fn target_subdir_for(profile: &str) -> &str {
     }
 }
 
+/// Phase 3b (2026-07-13): build a Lamport-stamped ref value using
+/// the runner's wall clock as the clock and `blake3(hostname)[..8]`
+/// as the node stamp. On concurrent PutRefVersioned calls, the later
+/// wall-clock write deterministically wins; ties are broken by the
+/// hostname hash.
+///
+/// Falls back to a zero-node stamp when `hostname()` fails
+/// (extremely rare on Unix), giving the wall clock alone as the
+/// merge key.
+fn build_stamped_ref(
+    blob_id: &BlobId,
+) -> crate::cluster::refs::StampedRef {
+    let clock = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .unwrap_or(0);
+    let node = match hostname_string() {
+        Some(h) => crate::cluster::refs::node_stamp_for(&h),
+        None => [0u8; 8],
+    };
+    crate::cluster::refs::StampedRef {
+        value: *blob_id.as_bytes(),
+        clock,
+        node,
+    }
+}
+
+/// Cheap `hostname` probe. Reads `/etc/hostname` on Linux, falls
+/// back to `HOSTNAME`/`COMPUTERNAME` env vars. `None` on any read
+/// failure — callers treat that as "no node identity available."
+fn hostname_string() -> Option<String> {
+    if let Ok(s) = std::fs::read_to_string("/etc/hostname") {
+        let trimmed = s.trim();
+        if !trimmed.is_empty() {
+            return Some(trimmed.to_string());
+        }
+    }
+    for var in ["HOSTNAME", "COMPUTERNAME"] {
+        if let Ok(v) = std::env::var(var) {
+            if !v.is_empty() {
+                return Some(v);
+            }
+        }
+    }
+    None
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

claw-store/src/cluster/rpc.rs

@@ -153,6 +153,11 @@ pub enum Method {
     /// Reply: 48 bytes on hit, single-byte
     /// [`ErrorCode::NotFound`] on miss.
     GetRefVersioned = 0x16,
+    /// Phase 3b (2026-07-13): strict local-only stamped-ref lookup.
+    /// Same wire shape as [`Method::GetRefVersioned`] but the peer
+    /// MUST NOT forward on miss. Used by daemons doing ref-forwarding
+    /// so they never loop.
+    GetRefVersionedLocal = 0x17,
 }
 
 impl Method {
@@ -182,6 +187,7 @@ impl Method {
             0x14 => Some(Method::GetRefLocal),
             0x15 => Some(Method::PutRefVersioned),
             0x16 => Some(Method::GetRefVersioned),
+            0x17 => Some(Method::GetRefVersionedLocal),
             _ => None,
         }
     }
@@ -658,6 +664,31 @@ impl RpcRouter {
                 }
             }
             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)),
+                };
+                // Local first.
+                if let Some(s) = store.get_stamped(&key).await? {
+                    self.metrics.record_get_ref_hit();
+                    return Ok(HandlerOutcome::Reply(s.to_bytes().to_vec()));
+                }
+                // Phase 3b: cross-runner sharing for stamped refs.
+                // On miss, fan out to peers; first hit pulls the blob
+                // locally + put_stamped so subsequent lookups are
+                // pure local hits (same semantics as GetRef path).
+                if let Some(s) = self.forward_get_ref_versioned(&key).await {
+                    self.metrics.record_get_ref_hit();
+                    return Ok(HandlerOutcome::Reply(s.to_bytes().to_vec()));
+                }
+                self.metrics.record_get_ref_miss();
+                Ok(HandlerOutcome::Error(ErrorCode::NotFound))
+            }
+            Method::GetRefVersionedLocal => {
                 let store = match &self.ref_store {
                     Some(s) => s,
                     None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
@@ -812,6 +843,73 @@ impl RpcRouter {
         }
         None
     }
+
+    /// Phase 3b (2026-07-13): stamped-ref forwarder. Same shape as
+    /// [`Self::forward_get_ref`] but uses `GetRefVersionedLocal` and
+    /// `put_stamped` on the local side so the CRDT-merge semantics
+    /// carry through cross-node lookups.
+    async fn forward_get_ref_versioned(
+        &self,
+        key: &RefKey,
+    ) -> Option<StampedRef> {
+        let client = self.outbound_client.as_ref()?.clone();
+        let blob_store = self.blob_store.as_ref()?.clone();
+        let ref_store = self.ref_store.as_ref()?.clone();
+        let peers: Vec<PeerView> = self
+            .gossip
+            .peers()
+            .await
+            .into_iter()
+            .filter(|p| p.alive && p.rpc_lan.or(p.rpc_tailscale).is_some())
+            .collect();
+        if peers.is_empty() {
+            return None;
+        }
+        let mut set: tokio::task::JoinSet<Option<StampedRef>> =
+            tokio::task::JoinSet::new();
+        let key_owned = *key;
+        for peer in peers {
+            let client = client.clone();
+            let blob_store = blob_store.clone();
+            let ref_store = ref_store.clone();
+            set.spawn(async move {
+                let addr = peer.rpc_lan.or(peer.rpc_tailscale)?;
+                let conn = match tokio::time::timeout(
+                    std::time::Duration::from_secs(3),
+                    client.connect(addr, &peer.name),
+                )
+                .await
+                {
+                    Ok(Ok(c)) => c,
+                    _ => return None,
+                };
+                let stamped = match call_get_ref_versioned_local(&conn, &key_owned).await {
+                    Ok(Some(s)) => s,
+                    _ => return None,
+                };
+                let blob_id = crate::cluster::blob::BlobId::from_bytes(stamped.value);
+                if pull_blob_locally(&conn, &blob_store, &blob_id)
+                    .await
+                    .is_err()
+                {
+                    return None;
+                }
+                // `put_stamped` merges: if we happened to race a
+                // concurrent local write, the higher (clock, node)
+                // wins on disk. Either way return what we fetched
+                // so the caller sees a hit.
+                let _ = ref_store.put_stamped(&key_owned, stamped).await;
+                Some(stamped)
+            });
+        }
+        while let Some(join) = set.join_next().await {
+            if let Ok(Some(s)) = join {
+                set.abort_all();
+                return Some(s);
+            }
+        }
+        None
+    }
 }
 
 /// Ref-forwarding helper (2026-07-13): fetch a blob's manifest and

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

@@ -514,7 +514,25 @@ 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?;
+    call_get_ref_versioned_inner(conn, key, Method::GetRefVersioned).await
+}
+
+/// Phase 3b (2026-07-13): strict local-only stamped-ref lookup —
+/// the peer MUST NOT recurse. Used by daemons doing ref-forwarding
+/// so they never loop.
+pub async fn call_get_ref_versioned_local(
+    conn: &Connection,
+    key: &crate::cluster::refs::RefKey,
+) -> Result<Option<crate::cluster::refs::StampedRef>> {
+    call_get_ref_versioned_inner(conn, key, Method::GetRefVersionedLocal).await
+}
+
+async fn call_get_ref_versioned_inner(
+    conn: &Connection,
+    key: &crate::cluster::refs::RefKey,
+    method: Method,
+) -> Result<Option<crate::cluster::refs::StampedRef>> {
+    let reply = rpc_call(conn, method, key).await?;
     if reply.len() == 1 {
         match decode_error(reply[0]) {
             Some(ErrorCode::NotFound) => return Ok(None),