claw-store/src/claw_cargo.rs

@@ -48,7 +48,7 @@ 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_get_ref_versioned, call_peer_status,
-    call_put_ref_versioned, prewarm_missing_chunks_between_parallel,
+    call_put_ref_versioned, call_put_tag_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};
@@ -184,6 +184,15 @@ struct PeerArgs {
     /// Workspace root. Defaults to the current directory.
     #[arg(long)]
     workspace: Option<PathBuf>,
+    /// Phase 3e (2026-07-13): opt-in namespace for the ref key. When
+    /// set, the fingerprint is combined with the namespace via
+    /// `blake3("clawstor.ns.v1" || namespace || fp)` before use — so
+    /// two runners on different namespaces (e.g. `clawverse/main` vs
+    /// `clawverse/pr-42`) don't collide on the same fingerprint.
+    /// Empty (the default) preserves the pre-3e behavior for
+    /// backward compat.
+    #[arg(long)]
+    namespace: Option<String>,
 }
 
 /// Args for the local-only `fingerprint` subcommand.
@@ -321,16 +330,15 @@ async fn connect_peer(cfg: &ResolvedClientConfig) -> Result<(QuicClient, quinn::
 /// garbage-collected blob).
 async fn peer_lookup(
     conn: &quinn::Connection,
-    fp: &Fingerprint,
+    key: &crate::cluster::refs::RefKey,
 ) -> Result<Option<(BlobId, crate::cluster::blob::BlobStat)>> {
-    let key = *fp.as_bytes();
     // Phase 3b (2026-07-13): read stamped refs first so cross-runner
     // ref-forwarding uses the CRDT-merge path. Fall back to legacy
     // 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? {
+        match call_get_ref_versioned(conn, key).await? {
             Some(s) => s.value,
-            None => match call_get_ref(conn, &key).await? {
+            None => match call_get_ref(conn, key).await? {
                 Some(v) => v,
                 None => return Ok(None),
             },
@@ -374,7 +382,8 @@ async fn cmd_status(args: PeerArgs) -> Result<()> {
     println!("tls:         {}", tls_dir.display());
 
     let (client, conn) = connect_peer(&resolved).await?;
-    match peer_lookup(&conn, &fp).await? {
+    let key = ref_key_for(&args.namespace, &fp);
+    match peer_lookup(&conn, &key).await? {
         Some((blob_id, stat)) => {
             println!("cache:       HIT");
             println!("  blob:      {}", blob_id);
@@ -412,7 +421,10 @@ async fn cmd_prefetch(args: PrefetchArgs) -> Result<()> {
                 return Ok(());
             }
         },
-        None => peer_lookup(&conn, &fp).await?,
+        None => {
+            let key = ref_key_for(&args.peer.namespace, &fp);
+            peer_lookup(&conn, &key).await?
+        }
     };
 
     let outcome = match lookup {
@@ -568,11 +580,20 @@ async fn cmd_build(args: BuildArgs) -> Result<()> {
         .join("target")
         .join(target_subdir_for(&resolved.profile));
     tracing::info!("fingerprint {}", fp);
+    let ref_key = ref_key_for(&args.peer.namespace, &fp);
+    if args.peer.namespace.as_deref().unwrap_or("").is_empty() {
+        tracing::info!("ref-key: fingerprint (no namespace)");
+    } else {
+        tracing::info!(
+            namespace = %args.peer.namespace.as_deref().unwrap_or(""),
+            "ref-key: namespaced fingerprint"
+        );
+    }
 
     let (client, conn) = connect_peer(&resolved).await?;
 
     let mut outcome = CacheOutcome::Miss;
-    match peer_lookup(&conn, &fp).await? {
+    match peer_lookup(&conn, &ref_key).await? {
         Some((blob_id, stat)) => {
             tracing::info!(
                 "cache HIT — blob {} ({} bytes, parallel={})",
@@ -675,7 +696,7 @@ async fn cmd_build(args: BuildArgs) -> Result<()> {
             // 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?;
+                call_put_ref_versioned(&conn, &ref_key, &stamped).await?;
             if merged {
                 tracing::info!(
                     "uploaded blob {} + set stamped ref (clock={})",
@@ -769,16 +790,39 @@ async fn cmd_pin(args: PinArgs) -> Result<()> {
         }
     };
 
-    // 2. Publish the tag → BlobId mapping.
+    // 2. Publish the tag → BlobId mapping. Phase 3c: use the stamped
-    call_put_tag(&conn, &args.name, blob_id.as_bytes()).await?;
+    // variant so concurrent `pin` calls race deterministically —
+    // higher (clock, node) wins, loser sees AlreadyExists.
+    let stamped_value = crate::cluster::tags::StampedTagValue {
+        value: *blob_id.as_bytes(),
+        clock: std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|d| d.as_secs())
+            .unwrap_or(0),
+        node: match hostname_string() {
+            Some(h) => crate::cluster::refs::node_stamp_for(&h),
+            None => [0u8; 8],
+        },
+    };
+    let merged =
+        call_put_tag_versioned(&conn, &args.name, &stamped_value).await?;
+    if !merged {
+        tracing::info!(
+            "another writer already published a dominant tag for {}; still ok",
+            args.name
+        );
+    }
 
-    // Field finding 2026-07-12: also stash the fingerprint under a
+    // Companion tag stashes the fingerprint bytes under
-    // companion tag `<name>.fingerprint`. Downstream nodes fed by
+    // `<name>.fingerprint` so prewarm downstream can PutRef locally.
-    // prewarm need to know the fingerprint to publish their own
-    // ref (`PutRef(fingerprint → blob)`) — otherwise a `build` on
-    // the same source misses even though the blob is present.
     let companion = fingerprint_companion_tag(&args.name);
-    call_put_tag(&conn, &companion, fp.as_bytes()).await?;
+    let companion_stamped = crate::cluster::tags::StampedTagValue {
+        value: *fp.as_bytes(),
+        clock: stamped_value.clock,
+        node: stamped_value.node,
+    };
+    let _ =
+        call_put_tag_versioned(&conn, &companion, &companion_stamped).await?;
 
     conn.close(quinn::VarInt::from_u32(0), b"done");
     client.shutdown().await;
@@ -1180,6 +1224,20 @@ fn build_stamped_ref(
     }
 }
 
+/// Phase 3e (2026-07-13): derive the ref-key to use for a lookup.
+/// Namespaced when the caller opted in via `--namespace` (or the
+/// layered config), otherwise the raw fingerprint bytes so the
+/// pre-3e default behavior is preserved.
+fn ref_key_for(
+    namespace: &Option<String>,
+    fp: &Fingerprint,
+) -> crate::cluster::refs::RefKey {
+    match namespace.as_deref().filter(|s| !s.is_empty()) {
+        Some(ns) => crate::cluster::refs::namespaced_ref_key(ns, fp.as_bytes()),
+        None => *fp.as_bytes(),
+    }
+}
+
 /// 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."

claw-store/src/cluster/refs.rs

@@ -99,6 +99,30 @@ pub fn node_stamp_for(name: &str) -> NodeStamp {
     out
 }
 
+/// Phase 3e (2026-07-13): derive a namespaced ref key from a
+/// `(namespace, fingerprint)` pair. Different namespaces produce
+/// different 32-byte keys for the same fingerprint — the primitive
+/// that lets a fleet segregate cache lookups by org/repo/branch/kind
+/// without touching the underlying content-addressed blob store.
+///
+/// Domain: `blake3("clawstor.ns.v1\0" || namespace || "\0" || fp)`.
+/// The version tag and the NUL delimiter make it collision-resistant
+/// against a future rewrite of this rule.
+///
+/// Empty namespace → the key is a deterministic function of the
+/// fingerprint alone, which callers can use for cluster-wide
+/// (default) lookups. Non-empty namespace → siloed lookups.
+pub fn namespaced_ref_key(namespace: &str, fingerprint: &[u8; 32]) -> RefKey {
+    let mut h = blake3::Hasher::new();
+    h.update(b"clawstor.ns.v1\0");
+    h.update(namespace.as_bytes());
+    h.update(b"\0");
+    h.update(fingerprint);
+    let mut out = [0u8; 32];
+    out.copy_from_slice(&h.finalize().as_bytes()[..32]);
+    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)]
@@ -420,6 +444,28 @@ mod tests {
         assert!(!c.dominates(&c));
     }
 
+    #[test]
+    fn namespaced_ref_key_is_stable_and_ns_scoped() {
+        // Phase 3e: the same (namespace, fingerprint) always maps to
+        // the same 32-byte key; different namespaces produce distinct
+        // keys for the same fingerprint; the empty namespace is a
+        // legitimate default lookup path.
+        let fp = [42u8; 32];
+        let a1 = namespaced_ref_key("clawverse/main", &fp);
+        let a2 = namespaced_ref_key("clawverse/main", &fp);
+        assert_eq!(a1, a2, "stable for same inputs");
+
+        let b = namespaced_ref_key("clawverse/pr-42", &fp);
+        assert_ne!(a1, b, "different namespaces silo the key");
+
+        let empty = namespaced_ref_key("", &fp);
+        assert_ne!(a1, empty, "empty namespace ≠ named namespace");
+        // Empty is a valid default and must not accidentally match
+        // the raw fingerprint (bypass would break the domain
+        // separation).
+        assert_ne!(empty, fp, "empty namespace still hashes the fp");
+    }
+
     #[test]
     fn node_stamp_is_stable_for_same_name() {
         let a = node_stamp_for("tank");

claw-store/src/cluster/rpc.rs

@@ -158,6 +158,19 @@ pub enum Method {
     /// MUST NOT forward on miss. Used by daemons doing ref-forwarding
     /// so they never loop.
     GetRefVersionedLocal = 0x17,
+    /// Phase 3c (2026-07-13): stamped/versioned PutTag. Merges under
+    /// the same `(clock, node)` total order as
+    /// [`Method::PutRefVersioned`].
+    /// `payload`: `key_len:u16 (LE) || key_bytes || stamped_value:48`.
+    /// Reply: single-byte status —
+    ///   `STREAM_STATUS_OK` = accepted (Merged);
+    ///   [`ErrorCode::AlreadyExists`] = rejected (older/equal).
+    PutTagVersioned = 0x18,
+    /// Phase 3c: fetch a stamped tag value.
+    /// `payload`: raw tag key bytes (variable length, up to 4 KiB).
+    /// Reply: 48 bytes on hit, single-byte
+    /// [`ErrorCode::NotFound`] on miss.
+    GetTagVersioned = 0x19,
 }
 
 impl Method {
@@ -188,6 +201,8 @@ impl Method {
             0x15 => Some(Method::PutRefVersioned),
             0x16 => Some(Method::GetRefVersioned),
             0x17 => Some(Method::GetRefVersionedLocal),
+            0x18 => Some(Method::PutTagVersioned),
+            0x19 => Some(Method::GetTagVersioned),
             _ => None,
         }
     }
@@ -739,6 +754,50 @@ impl RpcRouter {
                     }
                 }
             }
+            Method::PutTagVersioned => {
+                let store = match &self.tag_store {
+                    Some(s) => s,
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
+                };
+                // Wire: key_len:u16 (LE) || key_bytes || stamped_value:48.
+                let (key, stamped) =
+                    match crate::cluster::tags::decode_stamped_record(payload) {
+                        Ok(kv) => kv,
+                        Err(_) => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
+                    };
+                match store.put_stamped(&key, stamped).await {
+                    Ok(crate::cluster::tags::TagPutOutcome::Merged) => {
+                        Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK]))
+                    }
+                    Ok(crate::cluster::tags::TagPutOutcome::Rejected { .. }) => {
+                        Ok(HandlerOutcome::Error(ErrorCode::AlreadyExists))
+                    }
+                    Err(e) => {
+                        tracing::warn!(error = %e, "PutTagVersioned rejected");
+                        Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
+                    }
+                }
+            }
+            Method::GetTagVersioned => {
+                let store = match &self.tag_store {
+                    Some(s) => s,
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
+                };
+                let key = match std::str::from_utf8(payload) {
+                    Ok(s) if !s.is_empty() => s,
+                    _ => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
+                };
+                match store.get_stamped(key).await? {
+                    Some(s) => {
+                        self.metrics.record_get_tag_hit();
+                        Ok(HandlerOutcome::Reply(s.to_bytes().to_vec()))
+                    }
+                    None => {
+                        self.metrics.record_get_tag_miss();
+                        Ok(HandlerOutcome::Error(ErrorCode::NotFound))
+                    }
+                }
+            }
             Method::DeleteTag => {
                 let store = match &self.tag_store {
                     Some(s) => s,

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

@@ -546,6 +546,58 @@ async fn call_get_ref_versioned_inner(
     ))
 }
 
+// ── Phase 3c: stamped-tag client helpers ─────────────────────────────
+
+/// Phase 3c (2026-07-13): submit a stamped (CRDT-merge) PutTag.
+///
+/// Returns `Ok(true)` when the peer merged the write, `Ok(false)`
+/// when the peer rejected it because an equal-or-newer version
+/// already exists (`AlreadyExists`). Any other reply is an error.
+pub async fn call_put_tag_versioned(
+    conn: &Connection,
+    key: &str,
+    incoming: &crate::cluster::tags::StampedTagValue,
+) -> Result<bool> {
+    let payload = crate::cluster::tags::encode_stamped_record(key, incoming);
+    let reply = rpc_call(conn, Method::PutTagVersioned, &payload).await?;
+    if reply.len() != 1 {
+        bail!(
+            "expected single-byte PutTagVersioned 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 PutTagVersioned: {}", err.describe()),
+            None => bail!(
+                "peer replied with unknown byte 0x{:02x} for PutTagVersioned",
+                code
+            ),
+        },
+    }
+}
+
+/// Phase 3c: fetch a stamped tag value.
+pub async fn call_get_tag_versioned(
+    conn: &Connection,
+    key: &str,
+) -> Result<Option<crate::cluster::tags::StampedTagValue>> {
+    let reply = rpc_call(conn, Method::GetTagVersioned, key.as_bytes()).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::tags::StampedTagValue::from_bytes(&reply)
+            .context("decoding stamped tag reply")?,
+    ))
+}
+
 // ── Phase 5g: cache metrics client helper ────────────────────────────
 
 /// Fetch the peer's current cache-metrics snapshot.

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

@@ -1364,3 +1364,74 @@ async fn end_to_end_put_ref_versioned_merges_and_rejects() {
     tokio::time::sleep(Duration::from_millis(50)).await;
     acc.abort();
 }
+
+#[tokio::test]
+async fn end_to_end_put_tag_versioned_merges_and_rejects() {
+    // Phase 3c end-to-end: two writers publish stamped tags for the
+    // same key. Higher (clock, node) wins.
+    use crate::cluster::refs::node_stamp_for;
+    use crate::cluster::tags::StampedTagValue;
+
+    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
+    let (_tmp_a, router_a) = router_with_full_stack("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 = "clawverse:main:latest";
+    let node_x = node_stamp_for("runner-x");
+    let node_y = node_stamp_for("runner-y");
+
+    assert_eq!(call_get_tag_versioned(&conn, key).await.unwrap(), None);
+
+    let first = StampedTagValue {
+        value: [0x11; 32],
+        clock: 10,
+        node: node_x,
+    };
+    assert!(call_put_tag_versioned(&conn, key, &first).await.unwrap());
+    assert_eq!(
+        call_get_tag_versioned(&conn, key).await.unwrap(),
+        Some(first)
+    );
+
+    // Older clock → rejected.
+    let older = StampedTagValue {
+        value: [0x22; 32],
+        clock: 9,
+        node: node_y,
+    };
+    assert!(!call_put_tag_versioned(&conn, key, &older).await.unwrap());
+    assert_eq!(
+        call_get_tag_versioned(&conn, key).await.unwrap(),
+        Some(first)
+    );
+
+    // Higher clock always merges.
+    let latest = StampedTagValue {
+        value: [0x33; 32],
+        clock: 100,
+        node: node_x,
+    };
+    assert!(call_put_tag_versioned(&conn, key, &latest).await.unwrap());
+    assert_eq!(
+        call_get_tag_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();
+}

claw-store/src/cluster/tags.rs

@@ -31,6 +31,100 @@ use tokio::io::AsyncWriteExt;
 /// almost certainly a bug on the caller side.
 pub const MAX_TAG_KEY_BYTES: usize = 4096;
 
+/// Phase 3c (2026-07-13): the tag equivalent of
+/// [`crate::cluster::refs::StampedRef`]. Same `(clock, node)` total
+/// order, same idempotency semantics on the merge path.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct StampedTagValue {
+    pub value: [u8; 32],
+    pub clock: u64,
+    pub node: crate::cluster::refs::NodeStamp,
+}
+
+impl StampedTagValue {
+    /// 48-byte 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 tag value 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,
+        })
+    }
+
+    /// Same total-order semantics as [`crate::cluster::refs::StampedRef::dominates`].
+    pub fn stamp_dominates(&self, other: &Self) -> bool {
+        (self.clock, self.node) > (other.clock, other.node)
+    }
+}
+
+/// Merge outcome for [`TagStore::put_stamped`].
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum TagPutOutcome {
+    Merged,
+    Rejected { current: StampedTagValue },
+}
+
+/// Encode a stamped tag record as
+/// `key_len:u16 (LE) || key_bytes || stamped_value:48`.
+pub fn encode_stamped_record(key: &str, stamped: &StampedTagValue) -> Vec<u8> {
+    let key_bytes = key.as_bytes();
+    let mut out = Vec::with_capacity(2 + key_bytes.len() + StampedTagValue::ENCODED_LEN);
+    out.extend_from_slice(&(key_bytes.len() as u16).to_le_bytes());
+    out.extend_from_slice(key_bytes);
+    out.extend_from_slice(&stamped.to_bytes());
+    out
+}
+
+/// Reverse of [`encode_stamped_record`].
+pub fn decode_stamped_record(bytes: &[u8]) -> Result<(String, StampedTagValue)> {
+    if bytes.len() < 2 {
+        bail!(
+            "stamped tag record too short for length prefix ({} bytes)",
+            bytes.len()
+        );
+    }
+    let key_len = u16::from_le_bytes([bytes[0], bytes[1]]) as usize;
+    let expected = 2 + key_len + StampedTagValue::ENCODED_LEN;
+    if bytes.len() != expected {
+        bail!(
+            "stamped tag record length {} does not match declared shape \
+             (key_len={}, expected total {})",
+            bytes.len(),
+            key_len,
+            expected
+        );
+    }
+    let key_bytes = &bytes[2..2 + key_len];
+    let key = std::str::from_utf8(key_bytes)
+        .context("stamped tag key is not valid UTF-8")?
+        .to_string();
+    let stamped = StampedTagValue::from_bytes(&bytes[2 + key_len..])?;
+    Ok((key, stamped))
+}
+
 /// A tag entry as returned by `list`.
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub struct TagEntry {
@@ -182,6 +276,68 @@ impl TagStore {
             .join(format!("{hash_hex}.tag"))
     }
 
+    /// Phase 3c (2026-07-13): Lamport-stamped tag lookup. Reads from
+    /// `tags-v2/` — a separate namespace from raw `tags/` so the two
+    /// coexist during the cutover. Returns `None` when no stamped
+    /// value exists for this key.
+    pub async fn get_stamped(&self, key: &str) -> Result<Option<StampedTagValue>> {
+        validate_key(key)?;
+        let path = self.stamped_tag_path(key);
+        let bytes = match tokio::fs::read(&path).await {
+            Ok(b) => b,
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+            Err(e) => return Err(anyhow::Error::from(e)),
+        };
+        let (parsed_key, stamped) =
+            decode_stamped_record(&bytes).with_context(|| {
+                format!("decoding stamped tag record at {}", path.display())
+            })?;
+        if parsed_key != key {
+            bail!(
+                "stamped tag record at {} has key {:?} but was requested as {:?}",
+                path.display(),
+                parsed_key,
+                key
+            );
+        }
+        Ok(Some(stamped))
+    }
+
+    /// Phase 3c (2026-07-13): merge an incoming stamped tag. Higher
+    /// `(clock, node)` wins; equal is idempotent (returns
+    /// [`TagPutOutcome::Rejected`] with current).
+    ///
+    /// Never lowers the on-disk value.
+    pub async fn put_stamped(
+        &self,
+        key: &str,
+        incoming: StampedTagValue,
+    ) -> Result<TagPutOutcome> {
+        validate_key(key)?;
+        let path = self.stamped_tag_path(key);
+        if let Some(current) = self.get_stamped(key).await? {
+            if !incoming.stamp_dominates(&current) {
+                return Ok(TagPutOutcome::Rejected { current });
+            }
+        }
+        if let Some(parent) = path.parent() {
+            tokio::fs::create_dir_all(parent).await.with_context(|| {
+                format!("creating stamped-tag bucket {}", parent.display())
+            })?;
+        }
+        let bytes = encode_stamped_record(key, &incoming);
+        self.atomic_write(&path, &bytes).await?;
+        Ok(TagPutOutcome::Merged)
+    }
+
+    fn stamped_tag_path(&self, key: &str) -> PathBuf {
+        let hash_hex = hex32(blake3::hash(key.as_bytes()).as_bytes());
+        self.root
+            .join("tags-v2")
+            .join(&hash_hex[..2])
+            .join(format!("{hash_hex}.svtag"))
+    }
+
     async fn atomic_write(&self, final_path: &Path, bytes: &[u8]) -> Result<()> {
         let tmp_dir = self.root.join(".tmp");
         let tmp_name = format!(
@@ -430,4 +586,87 @@ mod tests {
         };
         assert!(entry.decode_value().is_err());
     }
+
+    // ── Phase 3c: stamped tags ───────────────────────────────────────
+
+    fn stamped_tag(value: u8, clock: u64, node: u8) -> StampedTagValue {
+        StampedTagValue {
+            value: [value; 32],
+            clock,
+            node: [node; 8],
+        }
+    }
+
+    #[test]
+    fn stamped_tag_encode_decode_round_trips() {
+        let s = stamped_tag(9, 7, 4);
+        let back = StampedTagValue::from_bytes(&s.to_bytes()).unwrap();
+        assert_eq!(back, s);
+    }
+
+    #[test]
+    fn stamped_record_encode_decode_round_trips() {
+        let s = stamped_tag(0xEE, 42, 3);
+        let bytes = encode_stamped_record("clawverse:main:latest", &s);
+        let (k, back) = decode_stamped_record(&bytes).unwrap();
+        assert_eq!(k, "clawverse:main:latest");
+        assert_eq!(back, s);
+    }
+
+    #[test]
+    fn stamped_record_rejects_truncated_input() {
+        assert!(decode_stamped_record(&[]).is_err());
+        assert!(decode_stamped_record(&[1u8]).is_err());
+        // Length prefix says 4 but total is 2+4+48=54 needed.
+        let bad = vec![4, 0, b'a', b'b'];
+        assert!(decode_stamped_record(&bad).is_err());
+    }
+
+    #[tokio::test]
+    async fn stamped_tag_put_get_round_trip() {
+        let (_tmp, store) = open();
+        let k = "clawverse:main:latest";
+        assert_eq!(store.get_stamped(k).await.unwrap(), None);
+        let s = stamped_tag(1, 1, 1);
+        assert_eq!(
+            store.put_stamped(k, s).await.unwrap(),
+            TagPutOutcome::Merged
+        );
+        assert_eq!(store.get_stamped(k).await.unwrap(), Some(s));
+    }
+
+    #[tokio::test]
+    async fn stamped_tag_merges_dominant_and_rejects_lower() {
+        let (_tmp, store) = open();
+        let k = "clawverse:main:latest";
+        let base = stamped_tag(1, 10, 5);
+        store.put_stamped(k, base).await.unwrap();
+
+        let older = stamped_tag(2, 9, 9);
+        assert!(matches!(
+            store.put_stamped(k, older).await.unwrap(),
+            TagPutOutcome::Rejected { current } if current == base
+        ));
+        assert_eq!(store.get_stamped(k).await.unwrap(), Some(base));
+
+        let higher = stamped_tag(3, 11, 0);
+        assert_eq!(
+            store.put_stamped(k, higher).await.unwrap(),
+            TagPutOutcome::Merged
+        );
+        assert_eq!(store.get_stamped(k).await.unwrap(), Some(higher));
+    }
+
+    #[tokio::test]
+    async fn stamped_and_unstamped_stores_are_independent() {
+        // put() writes to tags/, put_stamped() writes to tags-v2/.
+        // No clobber between the two.
+        let (_tmp, store) = open();
+        let k = "clawverse:main:latest";
+        store.put(k, &[0xAA; 32]).await.unwrap();
+        let s = stamped_tag(0xBB, 1, 1);
+        store.put_stamped(k, s).await.unwrap();
+        assert_eq!(store.get(k).await.unwrap(), Some([0xAA; 32]));
+        assert_eq!(store.get_stamped(k).await.unwrap(), Some(s));
+    }
 }