claw-store/src/claw_cargo.rs

@@ -46,7 +46,8 @@ use crate::cluster::build_cache::{
 };
 use crate::cluster::client_config::{ClientConfig, ResolvedClientConfig};
 use crate::cluster::rpc::{
-    call_blob_get_stream, call_blob_put_stream, call_blob_stat, call_get_ref, call_put_ref,
+    call_blob_get_stream, call_blob_put_stream, call_blob_stat, call_delete_tag, call_get_ref,
+    call_get_tag, call_list_tags, call_put_ref, call_put_tag,
 };
 use crate::cluster::transport::{NodeIdentity, QuicClient};
 
@@ -73,6 +74,33 @@ enum Cmd {
     Status(PeerArgs),
     /// Print the fingerprint only. Local, no network.
     Fingerprint(LocalArgs),
+    /// Pin the current fingerprint's cached BlobId under a human tag
+    /// like `clawverse:main:latest`. Requires the fingerprint to
+    /// already be in the peer's ref store (i.e. someone has built it).
+    Pin(PinArgs),
+    /// Delete a previously-pinned tag. The underlying BlobId is
+    /// untouched; only the human name goes away.
+    Unpin(UnpinArgs),
+    /// List every tag published on the peer.
+    ListTags(PeerArgs),
+}
+
+#[derive(clap::Args, Debug, Clone)]
+struct PinArgs {
+    /// Human-readable tag to publish (e.g. `clawverse:main:latest`).
+    #[arg(long)]
+    name: String,
+    #[command(flatten)]
+    peer: PeerArgs,
+}
+
+#[derive(clap::Args, Debug, Clone)]
+struct UnpinArgs {
+    /// Human-readable tag to delete.
+    #[arg(long)]
+    name: String,
+    #[command(flatten)]
+    peer: PeerArgs,
 }
 
 /// Args every network-touching subcommand accepts (with layered
@@ -134,6 +162,9 @@ async fn main() -> Result<()> {
         Cmd::Prefetch(args) => cmd_prefetch(args).await,
         Cmd::Status(args) => cmd_status(args).await,
         Cmd::Fingerprint(args) => cmd_fingerprint(args),
+        Cmd::Pin(args) => cmd_pin(args).await,
+        Cmd::Unpin(args) => cmd_unpin(args).await,
+        Cmd::ListTags(args) => cmd_list_tags(args).await,
     }
 }
 
@@ -450,6 +481,107 @@ fn print_summary(fp: &Fingerprint, outcome: &CacheOutcome, cargo_elapsed: std::t
     println!("────────────────────────────────────────────────────");
 }
 
+// ── pin / unpin / list-tags ──────────────────────────────────────────
+
+async fn cmd_pin(args: PinArgs) -> Result<()> {
+    let (_workspace, resolved, fp) = setup_peer(&args.peer)?;
+    let (client, conn) = connect_peer(&resolved).await?;
+
+    // 1. Resolve the fingerprint → BlobId via the ref store. Refuse
+    //    to pin something that isn't cached yet — otherwise the tag
+    //    would point at a value that no producer ever put there.
+    let key = *fp.as_bytes();
+    let blob_id = match call_get_ref(&conn, &key).await? {
+        Some(v) => BlobId::from_bytes(v),
+        None => {
+            conn.close(quinn::VarInt::from_u32(0), b"done");
+            client.shutdown().await;
+            anyhow::bail!(
+                "cannot pin: fingerprint {} not in peer's ref store yet — build first",
+                fp
+            );
+        }
+    };
+
+    // 2. Publish the tag → BlobId mapping.
+    call_put_tag(&conn, &args.name, blob_id.as_bytes()).await?;
+
+    conn.close(quinn::VarInt::from_u32(0), b"done");
+    client.shutdown().await;
+
+    println!("pinned:      {}", args.name);
+    println!("fingerprint: {}", fp);
+    println!("blob:        {}", blob_id);
+    Ok(())
+}
+
+async fn cmd_unpin(args: UnpinArgs) -> Result<()> {
+    let workspace = args
+        .peer
+        .workspace
+        .clone()
+        .unwrap_or_else(|| std::env::current_dir().expect("cwd"));
+    let layered = ClientConfig::load_layered(&workspace)?;
+    let resolved = ResolvedClientConfig::resolve(
+        layered,
+        args.peer.peer.clone(),
+        args.peer.peer_addr,
+        args.peer.tls_dir.clone(),
+        args.peer.profile.clone(),
+        args.peer.features.clone(),
+        "dev",
+    );
+    let (client, conn) = connect_peer(&resolved).await?;
+    let removed = call_delete_tag(&conn, &args.name).await?;
+    conn.close(quinn::VarInt::from_u32(0), b"done");
+    client.shutdown().await;
+
+    if removed {
+        println!("unpinned: {}", args.name);
+    } else {
+        println!("no such tag: {}", args.name);
+    }
+    Ok(())
+}
+
+async fn cmd_list_tags(args: PeerArgs) -> Result<()> {
+    let workspace = args
+        .workspace
+        .clone()
+        .unwrap_or_else(|| std::env::current_dir().expect("cwd"));
+    let layered = ClientConfig::load_layered(&workspace)?;
+    let resolved = ResolvedClientConfig::resolve(
+        layered,
+        args.peer.clone(),
+        args.peer_addr,
+        args.tls_dir.clone(),
+        args.profile.clone(),
+        args.features.clone(),
+        "dev",
+    );
+    let (client, conn) = connect_peer(&resolved).await?;
+    let tags = call_list_tags(&conn).await?;
+    conn.close(quinn::VarInt::from_u32(0), b"done");
+    client.shutdown().await;
+
+    if tags.is_empty() {
+        println!("(no tags)");
+    } else {
+        for entry in &tags {
+            println!("{}\t{}", entry.key, entry.value_hex);
+        }
+    }
+    Ok(())
+}
+
+// Suppress the unused-import warning when the CLI ends up not using
+// `call_get_tag` directly — it's still available for future subcommands
+// (e.g. `prefetch --pin <tag>`) and part of the public client API.
+#[allow(dead_code)]
+async fn _reserved_call_get_tag(conn: &quinn::Connection, key: &str) -> Result<Option<[u8; 32]>> {
+    call_get_tag(conn, key).await
+}
+
 fn run_cargo(
     workspace: &Path,
     profile: &str,

claw-store/src/cluster.rs

@@ -21,6 +21,7 @@ pub mod gossip;
 pub mod refs;
 pub mod rpc;
 pub mod services;
+pub mod tags;
 pub mod transport;
 
 use crate::config::PeerEntry;

claw-store/src/cluster/rpc.rs

@@ -28,6 +28,7 @@
 use crate::cluster::blob::{BlobId, BlobManifest, BlobStat, BlobStore, ChunkHash};
 use crate::cluster::gossip::{ClusterGossip, PeerView};
 use crate::cluster::refs::{RefKey, RefStore, RefValue};
+use crate::cluster::tags::{TagEntry, TagStore};
 use anyhow::{bail, Context, Result};
 use quinn::{Connection, ConnectionError};
 use serde::{Deserialize, Serialize};
@@ -110,6 +111,21 @@ pub enum Method {
     /// Reply: `STREAM_STATUS_OK` (1 byte) on success, or a
     /// single-byte error code.
     PutRef = 0x0e,
+    /// Phase 5d: set a named tag pointing at a 32-byte value.
+    /// `payload`: `key_len:u16 (LE) || key_bytes || value:32bytes`.
+    /// Reply: `STREAM_STATUS_OK` on success.
+    PutTag = 0x0f,
+    /// Phase 5d: fetch a named tag's value.
+    /// `payload`: raw key bytes (variable length, up to 4 KiB).
+    /// Reply: 32-byte value on hit, `NotFound` on miss.
+    GetTag = 0x10,
+    /// Phase 5d: delete a named tag.
+    /// `payload`: raw key bytes.
+    /// Reply: `STREAM_STATUS_OK` (deleted) or `NotFound`.
+    DeleteTag = 0x11,
+    /// Phase 5d: list all tags on the peer. `payload`: empty.
+    /// Reply: JSON `Vec<TagEntry>` sorted by key.
+    ListTags = 0x12,
 }
 
 impl Method {
@@ -131,6 +147,10 @@ impl Method {
             0x0c => Some(Method::PutManifest),
             0x0d => Some(Method::GetRef),
             0x0e => Some(Method::PutRef),
+            0x0f => Some(Method::PutTag),
+            0x10 => Some(Method::GetTag),
+            0x11 => Some(Method::DeleteTag),
+            0x12 => Some(Method::ListTags),
             _ => None,
         }
     }
@@ -217,6 +237,7 @@ pub struct RpcRouter {
     gossip: Arc<ClusterGossip>,
     blob_store: Option<Arc<BlobStore>>,
     ref_store: Option<Arc<RefStore>>,
+    tag_store: Option<Arc<TagStore>>,
     local_name: String,
     local_zone: String,
 }
@@ -235,6 +256,7 @@ impl RpcRouter {
             gossip,
             blob_store: None,
             ref_store: None,
+            tag_store: None,
             local_name,
             local_zone,
         }
@@ -254,6 +276,13 @@ impl RpcRouter {
         self
     }
 
+    /// Attach a local tag store (Phase 5d). Enables the
+    /// `PutTag` / `GetTag` / `DeleteTag` / `ListTags` methods.
+    pub fn with_tag_store(mut self, store: Arc<TagStore>) -> Self {
+        self.tag_store = Some(store);
+        self
+    }
+
     pub fn blob_store(&self) -> Option<&Arc<BlobStore>> {
         self.blob_store.as_ref()
     }
@@ -262,6 +291,10 @@ impl RpcRouter {
         self.ref_store.as_ref()
     }
 
+    pub fn tag_store(&self) -> Option<&Arc<TagStore>> {
+        self.tag_store.as_ref()
+    }
+
     /// Dispatch a single request. Called by [`serve_connection`] for
     /// every accepted bidi stream. Test code may call it directly to
     /// bypass the transport.
@@ -483,6 +516,62 @@ impl RpcRouter {
                 store.put(&key, &value).await?;
                 Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK]))
             }
+            Method::PutTag => {
+                let store = match &self.tag_store {
+                    Some(s) => s,
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
+                };
+                let (key, value) = match crate::cluster::tags::decode_record(payload) {
+                    Ok(kv) => kv,
+                    Err(_) => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
+                };
+                match store.put(&key, &value).await {
+                    Ok(()) => Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK])),
+                    Err(e) => {
+                        tracing::warn!(error = %e, "PutTag rejected");
+                        Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
+                    }
+                }
+            }
+            Method::GetTag => {
+                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(key).await? {
+                    Some(value) => Ok(HandlerOutcome::Reply(value.to_vec())),
+                    None => Ok(HandlerOutcome::Error(ErrorCode::NotFound)),
+                }
+            }
+            Method::DeleteTag => {
+                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)),
+                };
+                if store.delete(key).await? {
+                    Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK]))
+                } else {
+                    Ok(HandlerOutcome::Error(ErrorCode::NotFound))
+                }
+            }
+            Method::ListTags => {
+                let store = match &self.tag_store {
+                    Some(s) => s,
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
+                };
+                let entries = store.list().await?;
+                let json = serde_json::to_vec(&entries)
+                    .context("encoding TagEntry list as JSON")?;
+                Ok(HandlerOutcome::Reply(json))
+            }
         }
     }
 }
@@ -651,103 +740,6 @@ async fn handle_blob_get_stream(
     Ok(())
 }
 
-// ── Streaming client helpers ─────────────────────────────────────────
-
-/// Upload a blob by streaming from `reader` into the peer's store.
-/// Never buffers the whole blob — memory usage is dominated by
-/// tokio's default copy buffer. Returns the peer-assigned BlobId,
-/// which is NOT locally re-verified (the peer doesn't get the whole
-/// content in memory at once so we can't cheaply hash from our side);
-/// callers who need cryptographic proof of end-to-end integrity
-/// should follow up with `call_blob_stat` and compare hashes.
-pub async fn call_blob_put_stream<R>(
-    conn: &Connection,
-    mut reader: R,
-) -> Result<BlobId>
-where
-    R: AsyncRead + Unpin,
-{
-    let (mut send, mut recv) = conn
-        .open_bi()
-        .await
-        .context("opening bidi stream for BlobPutStream")?;
-    send.write_all(&[Method::BlobPutStream.as_byte()])
-        .await
-        .context("writing method tag")?;
-    tokio::io::copy(&mut reader, &mut send)
-        .await
-        .context("streaming blob content to peer")?;
-    send.finish().context("finishing BlobPutStream send")?;
-    let mut status = [0u8; 1];
-    recv.read_exact(&mut status)
-        .await
-        .context("reading BlobPutStream status byte")?;
-    if let Some(code) = decode_error(status[0]) {
-        bail!("peer replied with error: {}", code.describe());
-    }
-    if status[0] != STREAM_STATUS_OK {
-        bail!(
-            "peer replied with unknown status byte 0x{:02x}",
-            status[0]
-        );
-    }
-    let mut id_bytes = [0u8; 32];
-    recv.read_exact(&mut id_bytes)
-        .await
-        .context("reading BlobId from BlobPutStream reply")?;
-    Ok(BlobId::from_bytes(id_bytes))
-}
-
-/// Fetch a blob by streaming its bytes into `writer`. Returns
-/// `Ok(false)` when the peer replies [`ErrorCode::NotFound`]; other
-/// error codes surface as `Err`. `writer` receives exactly the blob
-/// content — no leading status byte, no framing.
-pub async fn call_blob_get_stream<W>(
-    conn: &Connection,
-    id: &BlobId,
-    writer: &mut W,
-) -> Result<bool>
-where
-    W: AsyncWrite + Unpin,
-{
-    let (mut send, mut recv) = conn
-        .open_bi()
-        .await
-        .context("opening bidi stream for BlobGetStream")?;
-    let mut req = Vec::with_capacity(33);
-    req.push(Method::BlobGetStream.as_byte());
-    req.extend_from_slice(id.as_bytes());
-    send.write_all(&req)
-        .await
-        .context("writing BlobGetStream request")?;
-    send.finish().context("finishing BlobGetStream send")?;
-
-    let mut status = [0u8; 1];
-    recv.read_exact(&mut status)
-        .await
-        .context("reading BlobGetStream status byte")?;
-    if let Some(code) = decode_error(status[0]) {
-        if code == ErrorCode::NotFound {
-            return Ok(false);
-        }
-        bail!("peer replied with error: {}", code.describe());
-    }
-    if status[0] != STREAM_STATUS_OK {
-        bail!(
-            "peer replied with unknown status byte 0x{:02x}",
-            status[0]
-        );
-    }
-    tokio::io::copy(&mut recv, writer)
-        .await
-        .context("streaming blob content from peer")?;
-    writer
-        .flush()
-        .await
-        .context("flushing destination after BlobGetStream")?;
-    Ok(true)
-}
-
 /// Turn a raw wire-format request into a response — either the
 /// router's real reply or a single-byte error code. Extracted so tests
 /// can hit it without a QUIC connection.
@@ -769,400 +761,12 @@ async fn dispatch(router: &RpcRouter, request: &[u8]) -> Vec<u8> {
     }
 }
 
-/// Client-side: open a bidi stream, write `method || payload`, read
+// Client helpers live in a submodule to stay under the 1300-line
-/// reply. Returns the raw reply bytes; callers deserialise per method.
+// ceiling on this file. Re-exported so external code keeps writing
-pub async fn rpc_call(
+// `cluster::rpc::call_*`.
-    conn: &Connection,
+#[path = "rpc/client.rs"]
-    method: Method,
+mod client;
-    payload: &[u8],
+pub use client::*;
-) -> Result<Vec<u8>> {
-    if payload.len() + 1 > MAX_MESSAGE_BYTES {
-        bail!(
-            "RPC payload {} bytes (+1 for method tag) exceeds cap {}",
-            payload.len(),
-            MAX_MESSAGE_BYTES
-        );
-    }
-    let (mut send, mut recv) = conn
-        .open_bi()
-        .await
-        .context("opening bidi stream for RPC")?;
-    let mut buf = Vec::with_capacity(1 + payload.len());
-    buf.push(method.as_byte());
-    buf.extend_from_slice(payload);
-    send.write_all(&buf).await.context("writing RPC request")?;
-    send.finish().context("finishing RPC send stream")?;
-    let reply = recv
-        .read_to_end(MAX_MESSAGE_BYTES)
-        .await
-        .context("reading RPC reply")?;
-    Ok(reply)
-}
-
-/// Convenience wrapper for [`Method::Ping`]. Sends `payload`, returns
-/// the peer's echo (`"pong:" || payload`) with the prefix stripped.
-pub async fn call_ping(conn: &Connection, payload: &[u8]) -> Result<Vec<u8>> {
-    let reply = rpc_call(conn, Method::Ping, payload).await?;
-    if reply.len() == 1 {
-        if let Some(code) = decode_error(reply[0]) {
-            bail!("peer replied with error: {}", code.describe());
-        }
-    }
-    if let Some(rest) = reply.strip_prefix(b"pong:") {
-        Ok(rest.to_vec())
-    } else {
-        bail!(
-            "peer replied with unexpected shape: {} bytes, no 'pong:' prefix",
-            reply.len()
-        );
-    }
-}
-
-/// Convenience wrapper for [`Method::PeerStatus`]. Sends an empty
-/// payload, deserialises the JSON response.
-pub async fn call_peer_status(conn: &Connection) -> Result<PeerStatusReply> {
-    let reply = rpc_call(conn, Method::PeerStatus, &[]).await?;
-    if reply.len() == 1 {
-        if let Some(code) = decode_error(reply[0]) {
-            bail!("peer replied with error: {}", code.describe());
-        }
-    }
-    serde_json::from_slice(&reply).context("decoding PeerStatusReply JSON")
-}
-
-/// Recognise a single-byte reply as one of our error codes. Returns
-/// `None` for any other single-byte value (which is a valid reply,
-/// just an unusually short one).
-pub fn decode_error(b: u8) -> Option<ErrorCode> {
-    match b {
-        0xf0 => Some(ErrorCode::EmptyRequest),
-        0xf1 => Some(ErrorCode::UnknownMethod),
-        0xf2 => Some(ErrorCode::HandlerFailure),
-        0xf3 => Some(ErrorCode::NotFound),
-        0xf4 => Some(ErrorCode::InvalidRequest),
-        0xf5 => Some(ErrorCode::NotConfigured),
-        _ => None,
-    }
-}
-
-// ── Blob RPC client helpers ───────────────────────────────────────────
-
-/// Ask the peer for a blob's size + chunk count. `Ok(None)` when the
-/// peer replies [`ErrorCode::NotFound`]; other error codes surface as
-/// `Err`.
-pub async fn call_blob_stat(
-    conn: &Connection,
-    id: &BlobId,
-) -> Result<Option<BlobStat>> {
-    let reply = rpc_call(conn, Method::BlobStat, id.as_bytes()).await?;
-    if reply.len() == 1 {
-        match decode_error(reply[0]) {
-            Some(ErrorCode::NotFound) => return Ok(None),
-            Some(code) => bail!("peer replied with error: {}", code.describe()),
-            None => {} // single-byte JSON like "1" is technically possible; fall through
-        }
-    }
-    let stat = serde_json::from_slice(&reply).context("decoding BlobStat JSON")?;
-    Ok(Some(stat))
-}
-
-/// Fetch a blob's raw bytes. `Ok(None)` when the peer replies
-/// [`ErrorCode::NotFound`]. Callers that need many-GB transfers should
-/// use the streaming variant (Phase 2c); this helper caps at
-/// [`MAX_MESSAGE_BYTES`].
-pub async fn call_blob_get(
-    conn: &Connection,
-    id: &BlobId,
-) -> Result<Option<Vec<u8>>> {
-    let reply = rpc_call(conn, Method::BlobGet, id.as_bytes()).await?;
-    if reply.len() == 1 {
-        match decode_error(reply[0]) {
-            Some(ErrorCode::NotFound) => return Ok(None),
-            Some(code) => bail!("peer replied with error: {}", code.describe()),
-            None => {} // single-byte blob is legitimate content
-        }
-    }
-    Ok(Some(reply))
-}
-
-/// Upload a blob to the peer's store. Returns the BlobId the peer
-/// assigned; must match the local hash of `bytes` (proves the peer
-/// stored what we sent).
-pub async fn call_blob_put(conn: &Connection, bytes: &[u8]) -> Result<BlobId> {
-    let reply = rpc_call(conn, Method::BlobPut, bytes).await?;
-    if reply.len() == 1 {
-        if let Some(code) = decode_error(reply[0]) {
-            bail!("peer replied with error: {}", code.describe());
-        }
-    }
-    if reply.len() != 32 {
-        bail!(
-            "expected 32-byte BlobId in reply, got {} bytes",
-            reply.len()
-        );
-    }
-    let mut buf = [0u8; 32];
-    buf.copy_from_slice(&reply);
-    let assigned = BlobId::from_bytes(buf);
-    let expected = BlobId::from_bytes(blake3::hash(bytes).into());
-    if assigned != expected {
-        bail!(
-            "peer returned BlobId {} but content hashes to {}; corruption or protocol drift",
-            assigned.to_hex(),
-            expected.to_hex()
-        );
-    }
-    Ok(assigned)
-}
-
-/// Fetch a blob's manifest (chunk-list + size). `Ok(None)` on NotFound.
-pub async fn call_blob_load_manifest(
-    conn: &Connection,
-    id: &BlobId,
-) -> Result<Option<BlobManifest>> {
-    let reply = rpc_call(conn, Method::BlobLoadManifest, id.as_bytes()).await?;
-    if reply.len() == 1 {
-        match decode_error(reply[0]) {
-            Some(ErrorCode::NotFound) => return Ok(None),
-            Some(code) => bail!("peer replied with error: {}", code.describe()),
-            None => {}
-        }
-    }
-    let manifest =
-        serde_json::from_slice(&reply).context("decoding BlobManifest JSON")?;
-    Ok(Some(manifest))
-}
-
-// ── Phase 2d: chunk-level client helpers ─────────────────────────────
-
-/// Does the peer already have this chunk? `Ok(true)` on presence,
-/// `Ok(false)` on NotFound. Other errors surface as `Err`.
-pub async fn call_has_chunk(
-    conn: &Connection,
-    hash: &ChunkHash,
-) -> Result<bool> {
-    let reply = rpc_call(conn, Method::HasChunk, hash.as_bytes()).await?;
-    if reply.len() != 1 {
-        bail!(
-            "expected single-byte HasChunk reply, got {} bytes",
-            reply.len()
-        );
-    }
-    match reply[0] {
-        STREAM_STATUS_OK => Ok(true),
-        code => match decode_error(code) {
-            Some(ErrorCode::NotFound) => Ok(false),
-            Some(err) => bail!("peer replied with error: {}", err.describe()),
-            None => bail!(
-                "peer replied with unknown byte 0x{:02x} for HasChunk",
-                code
-            ),
-        },
-    }
-}
-
-/// Upload a single chunk to the peer's store. Peer verifies bytes
-/// hash to `hash` before writing; a mismatch surfaces as `Err`.
-pub async fn call_put_chunk(
-    conn: &Connection,
-    hash: &ChunkHash,
-    bytes: &[u8],
-) -> Result<()> {
-    let mut payload = Vec::with_capacity(32 + bytes.len());
-    payload.extend_from_slice(hash.as_bytes());
-    payload.extend_from_slice(bytes);
-    let reply = rpc_call(conn, Method::PutChunk, &payload).await?;
-    if reply.len() != 1 {
-        bail!(
-            "expected single-byte PutChunk reply, got {} bytes",
-            reply.len()
-        );
-    }
-    match reply[0] {
-        STREAM_STATUS_OK => Ok(()),
-        code => match decode_error(code) {
-            Some(err) => bail!("peer rejected chunk: {}", err.describe()),
-            None => bail!(
-                "peer replied with unknown byte 0x{:02x} for PutChunk",
-                code
-            ),
-        },
-    }
-}
-
-/// Fetch a single chunk from the peer's store. `Ok(None)` on NotFound.
-/// Verifies the returned bytes hash to `hash` locally — corruption or
-/// protocol drift surfaces as `Err`.
-pub async fn call_get_chunk(
-    conn: &Connection,
-    hash: &ChunkHash,
-) -> Result<Option<Vec<u8>>> {
-    let reply = rpc_call(conn, Method::GetChunk, hash.as_bytes()).await?;
-    if reply.is_empty() {
-        bail!("empty GetChunk reply");
-    }
-    match reply[0] {
-        STREAM_STATUS_OK => {
-            let bytes = reply[1..].to_vec();
-            let recomputed = ChunkHash::from_bytes(blake3::hash(&bytes).into());
-            if recomputed != *hash {
-                bail!(
-                    "GetChunk hash mismatch: requested {}, got bytes hashing to {}",
-                    hash.to_hex(),
-                    recomputed.to_hex()
-                );
-            }
-            Ok(Some(bytes))
-        }
-        code => match decode_error(code) {
-            Some(ErrorCode::NotFound) => Ok(None),
-            Some(err) => bail!("peer replied with error: {}", err.describe()),
-            None => {
-                if reply.len() == 1 {
-                    bail!(
-                        "peer replied with unknown byte 0x{:02x} for GetChunk",
-                        code
-                    );
-                }
-                // A single legitimate content byte with value 0xf3 is
-                // technically distinguishable from NotFound because
-                // GetChunk always prefixes with STREAM_STATUS_OK. The
-                // unreachable branch stays as belt-and-braces.
-                let bytes = reply[1..].to_vec();
-                let recomputed = ChunkHash::from_bytes(blake3::hash(&bytes).into());
-                if recomputed != *hash {
-                    bail!("GetChunk hash mismatch (fallback path)");
-                }
-                Ok(Some(bytes))
-            }
-        },
-    }
-}
-
-/// Commit a manifest whose chunks the peer should already have.
-/// Returns the list of chunk hashes the peer is still missing — an
-/// empty list means the manifest was persisted; a non-empty list
-/// tells the caller which chunks to upload before retrying.
-pub async fn call_put_manifest(
-    conn: &Connection,
-    manifest: &BlobManifest,
-) -> Result<Vec<ChunkHash>> {
-    let payload =
-        serde_json::to_vec(manifest).context("encoding BlobManifest as JSON")?;
-    let reply = rpc_call(conn, Method::PutManifest, &payload).await?;
-    if reply.len() == 1 {
-        if let Some(err) = decode_error(reply[0]) {
-            bail!("peer replied with error: {}", err.describe());
-        }
-    }
-    let decoded: PutManifestReply = serde_json::from_slice(&reply)
-        .context("decoding PutManifestReply JSON")?;
-    if decoded.blob_id != manifest.blob_id {
-        bail!(
-            "peer echoed blob_id {} but we sent {}",
-            decoded.blob_id.to_hex(),
-            manifest.blob_id.to_hex()
-        );
-    }
-    Ok(decoded.missing)
-}
-
-// ── Phase 5b: reference-store client helpers ─────────────────────────
-
-/// Look up a 32-byte value by 32-byte key. `Ok(None)` on `NotFound`.
-pub async fn call_get_ref(
-    conn: &Connection,
-    key: &RefKey,
-) -> Result<Option<RefValue>> {
-    let reply = rpc_call(conn, Method::GetRef, 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 => {}
-        }
-    }
-    if reply.len() != 32 {
-        bail!("expected 32-byte ref value, got {} bytes", reply.len());
-    }
-    let mut value = [0u8; 32];
-    value.copy_from_slice(&reply);
-    Ok(Some(value))
-}
-
-/// Set a 32-byte value for a 32-byte key. Overwrites any prior value.
-pub async fn call_put_ref(
-    conn: &Connection,
-    key: &RefKey,
-    value: &RefValue,
-) -> Result<()> {
-    let mut payload = Vec::with_capacity(64);
-    payload.extend_from_slice(key);
-    payload.extend_from_slice(value);
-    let reply = rpc_call(conn, Method::PutRef, &payload).await?;
-    if reply.len() != 1 {
-        bail!("expected single-byte PutRef reply, got {} bytes", reply.len());
-    }
-    match reply[0] {
-        STREAM_STATUS_OK => Ok(()),
-        code => match decode_error(code) {
-            Some(err) => bail!("peer rejected PutRef: {}", err.describe()),
-            None => bail!(
-                "peer replied with unknown byte 0x{:02x} for PutRef",
-                code
-            ),
-        },
-    }
-}
-
-/// High-level partial-sync helper: replicate a local blob to a peer
-/// by uploading only chunks the peer is missing.
-///
-/// Flow:
-///   1. Load the local manifest.
-///   2. For each chunk, ask the peer if it already has it.
-///   3. Upload the missing chunks.
-///   4. Send the manifest so the peer commits.
-///
-/// Returns `(uploaded_chunks, total_chunks)` — the difference is what
-/// dedup saved. On a warm peer with an identical prior build both
-/// numbers are equal to `manifest.chunks.len()` minus the count that
-/// were already present, and only the missing-chunk bytes cross the
-/// wire.
-pub async fn push_blob_missing_chunks(
-    conn: &Connection,
-    local: &BlobStore,
-    id: &BlobId,
-) -> Result<(usize, usize)> {
-    let manifest = local
-        .load_manifest(id)
-        .await?
-        .with_context(|| format!("blob {} not present locally", id.to_hex()))?;
-    let total = manifest.chunks.len();
-    let mut uploaded = 0usize;
-    for hash in &manifest.chunks {
-        if !call_has_chunk(conn, hash).await? {
-            let bytes = local
-                .read_chunk(hash)
-                .await?
-                .with_context(|| format!("chunk {} missing locally", hash.to_hex()))?;
-            call_put_chunk(conn, hash, &bytes).await?;
-            uploaded += 1;
-        }
-    }
-    // Commit the manifest. `missing` MUST be empty by now — if the
-    // peer still reports missing chunks after we uploaded them, the
-    // most likely cause is a store crash on their side; surface as Err.
-    let still_missing = call_put_manifest(conn, &manifest).await?;
-    if !still_missing.is_empty() {
-        bail!(
-            "peer still reports {} missing chunks after we uploaded {} — retry",
-            still_missing.len(),
-            uploaded
-        );
-    }
-    Ok((uploaded, total))
-}
 
 #[cfg(test)]
 #[path = "rpc/tests.rs"]

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

@@ -0,0 +1,589 @@
+//! Client-side RPC helpers split out of rpc.rs to keep the parent
+//! module under the 1300-line ceiling. Every function here calls
+//! [`super::rpc_call`] or opens a bidi stream directly and follows
+//! the wire format documented on [`super::Method`].
+//!
+//! Re-exported through `pub use client::*;` in the parent so external
+//! callers keep the `cluster::rpc::call_*` paths they've been using.
+
+use super::*;
+
+// ── Streaming client helpers ─────────────────────────────────────────
+
+/// Upload a blob by streaming from `reader` into the peer's store.
+/// Never buffers the whole blob — memory usage is dominated by
+/// tokio's default copy buffer. Returns the peer-assigned BlobId,
+/// which is NOT locally re-verified (the peer doesn't get the whole
+/// content in memory at once so we can't cheaply hash from our side);
+/// callers who need cryptographic proof of end-to-end integrity
+/// should follow up with `call_blob_stat` and compare hashes.
+pub async fn call_blob_put_stream<R>(
+    conn: &Connection,
+    mut reader: R,
+) -> Result<BlobId>
+where
+    R: AsyncRead + Unpin,
+{
+    let (mut send, mut recv) = conn
+        .open_bi()
+        .await
+        .context("opening bidi stream for BlobPutStream")?;
+    send.write_all(&[Method::BlobPutStream.as_byte()])
+        .await
+        .context("writing method tag")?;
+    tokio::io::copy(&mut reader, &mut send)
+        .await
+        .context("streaming blob content to peer")?;
+    send.finish().context("finishing BlobPutStream send")?;
+    let mut status = [0u8; 1];
+    recv.read_exact(&mut status)
+        .await
+        .context("reading BlobPutStream status byte")?;
+    if let Some(code) = decode_error(status[0]) {
+        bail!("peer replied with error: {}", code.describe());
+    }
+    if status[0] != STREAM_STATUS_OK {
+        bail!(
+            "peer replied with unknown status byte 0x{:02x}",
+            status[0]
+        );
+    }
+    let mut id_bytes = [0u8; 32];
+    recv.read_exact(&mut id_bytes)
+        .await
+        .context("reading BlobId from BlobPutStream reply")?;
+    Ok(BlobId::from_bytes(id_bytes))
+}
+
+/// Fetch a blob by streaming its bytes into `writer`. Returns
+/// `Ok(false)` when the peer replies [`ErrorCode::NotFound`]; other
+/// error codes surface as `Err`. `writer` receives exactly the blob
+/// content — no leading status byte, no framing.
+pub async fn call_blob_get_stream<W>(
+    conn: &Connection,
+    id: &BlobId,
+    writer: &mut W,
+) -> Result<bool>
+where
+    W: AsyncWrite + Unpin,
+{
+    let (mut send, mut recv) = conn
+        .open_bi()
+        .await
+        .context("opening bidi stream for BlobGetStream")?;
+    let mut req = Vec::with_capacity(33);
+    req.push(Method::BlobGetStream.as_byte());
+    req.extend_from_slice(id.as_bytes());
+    send.write_all(&req)
+        .await
+        .context("writing BlobGetStream request")?;
+    send.finish().context("finishing BlobGetStream send")?;
+
+    let mut status = [0u8; 1];
+    recv.read_exact(&mut status)
+        .await
+        .context("reading BlobGetStream status byte")?;
+    if let Some(code) = decode_error(status[0]) {
+        if code == ErrorCode::NotFound {
+            return Ok(false);
+        }
+        bail!("peer replied with error: {}", code.describe());
+    }
+    if status[0] != STREAM_STATUS_OK {
+        bail!(
+            "peer replied with unknown status byte 0x{:02x}",
+            status[0]
+        );
+    }
+    tokio::io::copy(&mut recv, writer)
+        .await
+        .context("streaming blob content from peer")?;
+    writer
+        .flush()
+        .await
+        .context("flushing destination after BlobGetStream")?;
+    Ok(true)
+}
+
+// `dispatch` is a server-side helper — it lives back in rpc.rs alongside
+// `serve_connection`. Tests reach it via `super::dispatch`.
+
+/// Client-side: open a bidi stream, write `method || payload`, read
+/// reply. Returns the raw reply bytes; callers deserialise per method.
+pub async fn rpc_call(
+    conn: &Connection,
+    method: Method,
+    payload: &[u8],
+) -> Result<Vec<u8>> {
+    if payload.len() + 1 > MAX_MESSAGE_BYTES {
+        bail!(
+            "RPC payload {} bytes (+1 for method tag) exceeds cap {}",
+            payload.len(),
+            MAX_MESSAGE_BYTES
+        );
+    }
+    let (mut send, mut recv) = conn
+        .open_bi()
+        .await
+        .context("opening bidi stream for RPC")?;
+    let mut buf = Vec::with_capacity(1 + payload.len());
+    buf.push(method.as_byte());
+    buf.extend_from_slice(payload);
+    send.write_all(&buf).await.context("writing RPC request")?;
+    send.finish().context("finishing RPC send stream")?;
+    let reply = recv
+        .read_to_end(MAX_MESSAGE_BYTES)
+        .await
+        .context("reading RPC reply")?;
+    Ok(reply)
+}
+
+/// Convenience wrapper for [`Method::Ping`]. Sends `payload`, returns
+/// the peer's echo (`"pong:" || payload`) with the prefix stripped.
+pub async fn call_ping(conn: &Connection, payload: &[u8]) -> Result<Vec<u8>> {
+    let reply = rpc_call(conn, Method::Ping, payload).await?;
+    if reply.len() == 1 {
+        if let Some(code) = decode_error(reply[0]) {
+            bail!("peer replied with error: {}", code.describe());
+        }
+    }
+    if let Some(rest) = reply.strip_prefix(b"pong:") {
+        Ok(rest.to_vec())
+    } else {
+        bail!(
+            "peer replied with unexpected shape: {} bytes, no 'pong:' prefix",
+            reply.len()
+        );
+    }
+}
+
+/// Convenience wrapper for [`Method::PeerStatus`]. Sends an empty
+/// payload, deserialises the JSON response.
+pub async fn call_peer_status(conn: &Connection) -> Result<PeerStatusReply> {
+    let reply = rpc_call(conn, Method::PeerStatus, &[]).await?;
+    if reply.len() == 1 {
+        if let Some(code) = decode_error(reply[0]) {
+            bail!("peer replied with error: {}", code.describe());
+        }
+    }
+    serde_json::from_slice(&reply).context("decoding PeerStatusReply JSON")
+}
+
+/// Recognise a single-byte reply as one of our error codes. Returns
+/// `None` for any other single-byte value (which is a valid reply,
+/// just an unusually short one).
+pub fn decode_error(b: u8) -> Option<ErrorCode> {
+    match b {
+        0xf0 => Some(ErrorCode::EmptyRequest),
+        0xf1 => Some(ErrorCode::UnknownMethod),
+        0xf2 => Some(ErrorCode::HandlerFailure),
+        0xf3 => Some(ErrorCode::NotFound),
+        0xf4 => Some(ErrorCode::InvalidRequest),
+        0xf5 => Some(ErrorCode::NotConfigured),
+        _ => None,
+    }
+}
+
+// ── Blob RPC client helpers ───────────────────────────────────────────
+
+/// Ask the peer for a blob's size + chunk count. `Ok(None)` when the
+/// peer replies [`ErrorCode::NotFound`]; other error codes surface as
+/// `Err`.
+pub async fn call_blob_stat(
+    conn: &Connection,
+    id: &BlobId,
+) -> Result<Option<BlobStat>> {
+    let reply = rpc_call(conn, Method::BlobStat, id.as_bytes()).await?;
+    if reply.len() == 1 {
+        match decode_error(reply[0]) {
+            Some(ErrorCode::NotFound) => return Ok(None),
+            Some(code) => bail!("peer replied with error: {}", code.describe()),
+            None => {} // single-byte JSON like "1" is technically possible; fall through
+        }
+    }
+    let stat = serde_json::from_slice(&reply).context("decoding BlobStat JSON")?;
+    Ok(Some(stat))
+}
+
+/// Fetch a blob's raw bytes. `Ok(None)` when the peer replies
+/// [`ErrorCode::NotFound`]. Callers that need many-GB transfers should
+/// use the streaming variant (Phase 2c); this helper caps at
+/// [`MAX_MESSAGE_BYTES`].
+pub async fn call_blob_get(
+    conn: &Connection,
+    id: &BlobId,
+) -> Result<Option<Vec<u8>>> {
+    let reply = rpc_call(conn, Method::BlobGet, id.as_bytes()).await?;
+    if reply.len() == 1 {
+        match decode_error(reply[0]) {
+            Some(ErrorCode::NotFound) => return Ok(None),
+            Some(code) => bail!("peer replied with error: {}", code.describe()),
+            None => {} // single-byte blob is legitimate content
+        }
+    }
+    Ok(Some(reply))
+}
+
+/// Upload a blob to the peer's store. Returns the BlobId the peer
+/// assigned; must match the local hash of `bytes` (proves the peer
+/// stored what we sent).
+pub async fn call_blob_put(conn: &Connection, bytes: &[u8]) -> Result<BlobId> {
+    let reply = rpc_call(conn, Method::BlobPut, bytes).await?;
+    if reply.len() == 1 {
+        if let Some(code) = decode_error(reply[0]) {
+            bail!("peer replied with error: {}", code.describe());
+        }
+    }
+    if reply.len() != 32 {
+        bail!(
+            "expected 32-byte BlobId in reply, got {} bytes",
+            reply.len()
+        );
+    }
+    let mut buf = [0u8; 32];
+    buf.copy_from_slice(&reply);
+    let assigned = BlobId::from_bytes(buf);
+    let expected = BlobId::from_bytes(blake3::hash(bytes).into());
+    if assigned != expected {
+        bail!(
+            "peer returned BlobId {} but content hashes to {}; corruption or protocol drift",
+            assigned.to_hex(),
+            expected.to_hex()
+        );
+    }
+    Ok(assigned)
+}
+
+/// Fetch a blob's manifest (chunk-list + size). `Ok(None)` on NotFound.
+pub async fn call_blob_load_manifest(
+    conn: &Connection,
+    id: &BlobId,
+) -> Result<Option<BlobManifest>> {
+    let reply = rpc_call(conn, Method::BlobLoadManifest, id.as_bytes()).await?;
+    if reply.len() == 1 {
+        match decode_error(reply[0]) {
+            Some(ErrorCode::NotFound) => return Ok(None),
+            Some(code) => bail!("peer replied with error: {}", code.describe()),
+            None => {}
+        }
+    }
+    let manifest =
+        serde_json::from_slice(&reply).context("decoding BlobManifest JSON")?;
+    Ok(Some(manifest))
+}
+
+// ── Phase 2d: chunk-level client helpers ─────────────────────────────
+
+/// Does the peer already have this chunk? `Ok(true)` on presence,
+/// `Ok(false)` on NotFound. Other errors surface as `Err`.
+pub async fn call_has_chunk(
+    conn: &Connection,
+    hash: &ChunkHash,
+) -> Result<bool> {
+    let reply = rpc_call(conn, Method::HasChunk, hash.as_bytes()).await?;
+    if reply.len() != 1 {
+        bail!(
+            "expected single-byte HasChunk reply, got {} bytes",
+            reply.len()
+        );
+    }
+    match reply[0] {
+        STREAM_STATUS_OK => Ok(true),
+        code => match decode_error(code) {
+            Some(ErrorCode::NotFound) => Ok(false),
+            Some(err) => bail!("peer replied with error: {}", err.describe()),
+            None => bail!(
+                "peer replied with unknown byte 0x{:02x} for HasChunk",
+                code
+            ),
+        },
+    }
+}
+
+/// Upload a single chunk to the peer's store. Peer verifies bytes
+/// hash to `hash` before writing; a mismatch surfaces as `Err`.
+pub async fn call_put_chunk(
+    conn: &Connection,
+    hash: &ChunkHash,
+    bytes: &[u8],
+) -> Result<()> {
+    let mut payload = Vec::with_capacity(32 + bytes.len());
+    payload.extend_from_slice(hash.as_bytes());
+    payload.extend_from_slice(bytes);
+    let reply = rpc_call(conn, Method::PutChunk, &payload).await?;
+    if reply.len() != 1 {
+        bail!(
+            "expected single-byte PutChunk reply, got {} bytes",
+            reply.len()
+        );
+    }
+    match reply[0] {
+        STREAM_STATUS_OK => Ok(()),
+        code => match decode_error(code) {
+            Some(err) => bail!("peer rejected chunk: {}", err.describe()),
+            None => bail!(
+                "peer replied with unknown byte 0x{:02x} for PutChunk",
+                code
+            ),
+        },
+    }
+}
+
+/// Fetch a single chunk from the peer's store. `Ok(None)` on NotFound.
+/// Verifies the returned bytes hash to `hash` locally — corruption or
+/// protocol drift surfaces as `Err`.
+pub async fn call_get_chunk(
+    conn: &Connection,
+    hash: &ChunkHash,
+) -> Result<Option<Vec<u8>>> {
+    let reply = rpc_call(conn, Method::GetChunk, hash.as_bytes()).await?;
+    if reply.is_empty() {
+        bail!("empty GetChunk reply");
+    }
+    match reply[0] {
+        STREAM_STATUS_OK => {
+            let bytes = reply[1..].to_vec();
+            let recomputed = ChunkHash::from_bytes(blake3::hash(&bytes).into());
+            if recomputed != *hash {
+                bail!(
+                    "GetChunk hash mismatch: requested {}, got bytes hashing to {}",
+                    hash.to_hex(),
+                    recomputed.to_hex()
+                );
+            }
+            Ok(Some(bytes))
+        }
+        code => match decode_error(code) {
+            Some(ErrorCode::NotFound) => Ok(None),
+            Some(err) => bail!("peer replied with error: {}", err.describe()),
+            None => {
+                if reply.len() == 1 {
+                    bail!(
+                        "peer replied with unknown byte 0x{:02x} for GetChunk",
+                        code
+                    );
+                }
+                // A single legitimate content byte with value 0xf3 is
+                // technically distinguishable from NotFound because
+                // GetChunk always prefixes with STREAM_STATUS_OK. The
+                // unreachable branch stays as belt-and-braces.
+                let bytes = reply[1..].to_vec();
+                let recomputed = ChunkHash::from_bytes(blake3::hash(&bytes).into());
+                if recomputed != *hash {
+                    bail!("GetChunk hash mismatch (fallback path)");
+                }
+                Ok(Some(bytes))
+            }
+        },
+    }
+}
+
+/// Commit a manifest whose chunks the peer should already have.
+/// Returns the list of chunk hashes the peer is still missing — an
+/// empty list means the manifest was persisted; a non-empty list
+/// tells the caller which chunks to upload before retrying.
+pub async fn call_put_manifest(
+    conn: &Connection,
+    manifest: &BlobManifest,
+) -> Result<Vec<ChunkHash>> {
+    let payload =
+        serde_json::to_vec(manifest).context("encoding BlobManifest as JSON")?;
+    let reply = rpc_call(conn, Method::PutManifest, &payload).await?;
+    if reply.len() == 1 {
+        if let Some(err) = decode_error(reply[0]) {
+            bail!("peer replied with error: {}", err.describe());
+        }
+    }
+    let decoded: PutManifestReply = serde_json::from_slice(&reply)
+        .context("decoding PutManifestReply JSON")?;
+    if decoded.blob_id != manifest.blob_id {
+        bail!(
+            "peer echoed blob_id {} but we sent {}",
+            decoded.blob_id.to_hex(),
+            manifest.blob_id.to_hex()
+        );
+    }
+    Ok(decoded.missing)
+}
+
+// ── Phase 5b: reference-store client helpers ─────────────────────────
+
+/// Look up a 32-byte value by 32-byte key. `Ok(None)` on `NotFound`.
+pub async fn call_get_ref(
+    conn: &Connection,
+    key: &RefKey,
+) -> Result<Option<RefValue>> {
+    let reply = rpc_call(conn, Method::GetRef, 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 => {}
+        }
+    }
+    if reply.len() != 32 {
+        bail!("expected 32-byte ref value, got {} bytes", reply.len());
+    }
+    let mut value = [0u8; 32];
+    value.copy_from_slice(&reply);
+    Ok(Some(value))
+}
+
+/// Set a 32-byte value for a 32-byte key. Overwrites any prior value.
+pub async fn call_put_ref(
+    conn: &Connection,
+    key: &RefKey,
+    value: &RefValue,
+) -> Result<()> {
+    let mut payload = Vec::with_capacity(64);
+    payload.extend_from_slice(key);
+    payload.extend_from_slice(value);
+    let reply = rpc_call(conn, Method::PutRef, &payload).await?;
+    if reply.len() != 1 {
+        bail!("expected single-byte PutRef reply, got {} bytes", reply.len());
+    }
+    match reply[0] {
+        STREAM_STATUS_OK => Ok(()),
+        code => match decode_error(code) {
+            Some(err) => bail!("peer rejected PutRef: {}", err.describe()),
+            None => bail!(
+                "peer replied with unknown byte 0x{:02x} for PutRef",
+                code
+            ),
+        },
+    }
+}
+
+// ── Phase 5d: tag-store client helpers ───────────────────────────────
+
+/// Publish (or overwrite) a named tag pointing at a 32-byte value.
+pub async fn call_put_tag(
+    conn: &Connection,
+    key: &str,
+    value: &[u8; 32],
+) -> Result<()> {
+    let payload = crate::cluster::tags::encode_record(key, value);
+    let reply = rpc_call(conn, Method::PutTag, &payload).await?;
+    if reply.len() != 1 {
+        bail!("expected single-byte PutTag reply, got {} bytes", reply.len());
+    }
+    match reply[0] {
+        STREAM_STATUS_OK => Ok(()),
+        code => match decode_error(code) {
+            Some(err) => bail!("peer rejected PutTag: {}", err.describe()),
+            None => bail!(
+                "peer replied with unknown byte 0x{:02x} for PutTag",
+                code
+            ),
+        },
+    }
+}
+
+/// Look up a named tag. `Ok(None)` on `NotFound`.
+pub async fn call_get_tag(
+    conn: &Connection,
+    key: &str,
+) -> Result<Option<[u8; 32]>> {
+    if key.is_empty() {
+        bail!("tag key cannot be empty");
+    }
+    let reply = rpc_call(conn, Method::GetTag, 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 => {}
+        }
+    }
+    if reply.len() != 32 {
+        bail!("expected 32-byte tag value, got {} bytes", reply.len());
+    }
+    let mut value = [0u8; 32];
+    value.copy_from_slice(&reply);
+    Ok(Some(value))
+}
+
+/// Delete a named tag. `Ok(false)` when no such tag existed.
+pub async fn call_delete_tag(conn: &Connection, key: &str) -> Result<bool> {
+    if key.is_empty() {
+        bail!("tag key cannot be empty");
+    }
+    let reply = rpc_call(conn, Method::DeleteTag, key.as_bytes()).await?;
+    if reply.len() != 1 {
+        bail!(
+            "expected single-byte DeleteTag reply, got {} bytes",
+            reply.len()
+        );
+    }
+    match reply[0] {
+        STREAM_STATUS_OK => Ok(true),
+        code => match decode_error(code) {
+            Some(ErrorCode::NotFound) => Ok(false),
+            Some(err) => bail!("peer replied with error: {}", err.describe()),
+            None => bail!(
+                "peer replied with unknown byte 0x{:02x} for DeleteTag",
+                code
+            ),
+        },
+    }
+}
+
+/// List every tag stored on the peer. Sorted by key.
+pub async fn call_list_tags(conn: &Connection) -> Result<Vec<TagEntry>> {
+    let reply = rpc_call(conn, Method::ListTags, &[]).await?;
+    if reply.len() == 1 {
+        if let Some(err) = decode_error(reply[0]) {
+            bail!("peer replied with error: {}", err.describe());
+        }
+    }
+    serde_json::from_slice(&reply).context("decoding TagEntry list JSON")
+}
+
+/// High-level partial-sync helper: replicate a local blob to a peer
+/// by uploading only chunks the peer is missing.
+///
+/// Flow:
+///   1. Load the local manifest.
+///   2. For each chunk, ask the peer if it already has it.
+///   3. Upload the missing chunks.
+///   4. Send the manifest so the peer commits.
+///
+/// Returns `(uploaded_chunks, total_chunks)` — the difference is what
+/// dedup saved. On a warm peer with an identical prior build both
+/// numbers are equal to `manifest.chunks.len()` minus the count that
+/// were already present, and only the missing-chunk bytes cross the
+/// wire.
+pub async fn push_blob_missing_chunks(
+    conn: &Connection,
+    local: &BlobStore,
+    id: &BlobId,
+) -> Result<(usize, usize)> {
+    let manifest = local
+        .load_manifest(id)
+        .await?
+        .with_context(|| format!("blob {} not present locally", id.to_hex()))?;
+    let total = manifest.chunks.len();
+    let mut uploaded = 0usize;
+    for hash in &manifest.chunks {
+        if !call_has_chunk(conn, hash).await? {
+            let bytes = local
+                .read_chunk(hash)
+                .await?
+                .with_context(|| format!("chunk {} missing locally", hash.to_hex()))?;
+            call_put_chunk(conn, hash, &bytes).await?;
+            uploaded += 1;
+        }
+    }
+    // Commit the manifest. `missing` MUST be empty by now — if the
+    // peer still reports missing chunks after we uploaded them, the
+    // most likely cause is a store crash on their side; surface as Err.
+    let still_missing = call_put_manifest(conn, &manifest).await?;
+    if !still_missing.is_empty() {
+        bail!(
+            "peer still reports {} missing chunks after we uploaded {} — retry",
+            still_missing.len(),
+            uploaded
+        );
+    }
+    Ok((uploaded, total))
+}

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

@@ -858,6 +858,168 @@ async fn end_to_end_put_ref_get_ref_over_real_quic() {
     accept_task.abort();
 }
 
+// ── Phase 5d: tag-store RPC ──────────────────────────────────────────
+
+async fn router_with_full_stack(name: &str, port: u16) -> (tempfile::TempDir, Arc<RpcRouter>) {
+    use crate::cluster::refs::RefStore;
+    use crate::cluster::tags::TagStore;
+    let gossip = bootstrap_gossip(name, port).await;
+    let tmp = tempfile::TempDir::new().unwrap();
+    let blob_store =
+        Arc::new(crate::cluster::blob::BlobStore::open(tmp.path().join("blobs")).unwrap());
+    let ref_store = Arc::new(RefStore::open(tmp.path().join("refs-db")).unwrap());
+    let tag_store = Arc::new(TagStore::open(tmp.path().join("tags-db")).unwrap());
+    let router = Arc::new(
+        RpcRouter::new(gossip, name.into(), "fabric-10g".into())
+            .with_blob_store(blob_store)
+            .with_ref_store(ref_store)
+            .with_tag_store(tag_store),
+    );
+    (tmp, router)
+}
+
+#[test]
+fn phase_5d_method_byte_encoding() {
+    assert_eq!(Method::PutTag.as_byte(), 0x0f);
+    assert_eq!(Method::GetTag.as_byte(), 0x10);
+    assert_eq!(Method::DeleteTag.as_byte(), 0x11);
+    assert_eq!(Method::ListTags.as_byte(), 0x12);
+    for m in [Method::PutTag, Method::GetTag, Method::DeleteTag, Method::ListTags] {
+        assert_eq!(Method::from_byte(m.as_byte()), Some(m));
+    }
+}
+
+#[tokio::test]
+async fn tag_rpcs_return_not_configured_without_store() {
+    let gossip = bootstrap_gossip("solo", next_port()).await;
+    let router = RpcRouter::new(gossip, "solo".into(), "z".into());
+    for method in [Method::PutTag, Method::GetTag, Method::DeleteTag, Method::ListTags] {
+        let mut req = vec![method.as_byte()];
+        req.extend_from_slice(b"any-key");
+        let reply = dispatch(&router, &req).await;
+        assert_eq!(
+            reply,
+            vec![ErrorCode::NotConfigured.as_byte()],
+            "method {method:?} should be NotConfigured"
+        );
+    }
+}
+
+#[tokio::test]
+async fn put_tag_stores_and_get_tag_reads_back() {
+    let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
+    let key = "clawverse:main:latest";
+    let value = [0x77u8; 32];
+
+    let put_payload = crate::cluster::tags::encode_record(key, &value);
+    let mut put_req = vec![Method::PutTag.as_byte()];
+    put_req.extend_from_slice(&put_payload);
+    assert_eq!(dispatch(&router, &put_req).await, vec![STREAM_STATUS_OK]);
+
+    let mut get_req = vec![Method::GetTag.as_byte()];
+    get_req.extend_from_slice(key.as_bytes());
+    assert_eq!(dispatch(&router, &get_req).await, value.to_vec());
+}
+
+#[tokio::test]
+async fn get_tag_returns_not_found_for_missing() {
+    let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
+    let mut req = vec![Method::GetTag.as_byte()];
+    req.extend_from_slice(b"never-set");
+    assert_eq!(
+        dispatch(&router, &req).await,
+        vec![ErrorCode::NotFound.as_byte()]
+    );
+}
+
+#[tokio::test]
+async fn get_tag_rejects_empty_key() {
+    let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
+    let req = vec![Method::GetTag.as_byte()]; // empty payload
+    assert_eq!(
+        dispatch(&router, &req).await,
+        vec![ErrorCode::InvalidRequest.as_byte()]
+    );
+}
+
+#[tokio::test]
+async fn delete_tag_removes_and_returns_not_found_after() {
+    let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
+    let store = router.tag_store().unwrap().clone();
+    store.put("removable", &[0u8; 32]).await.unwrap();
+
+    let mut req = vec![Method::DeleteTag.as_byte()];
+    req.extend_from_slice(b"removable");
+    assert_eq!(dispatch(&router, &req).await, vec![STREAM_STATUS_OK]);
+
+    // Second delete → NotFound.
+    assert_eq!(
+        dispatch(&router, &req).await,
+        vec![ErrorCode::NotFound.as_byte()]
+    );
+}
+
+#[tokio::test]
+async fn list_tags_returns_json_sorted() {
+    let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
+    let store = router.tag_store().unwrap().clone();
+    store.put("bravo", &[2u8; 32]).await.unwrap();
+    store.put("alpha", &[1u8; 32]).await.unwrap();
+
+    let req = vec![Method::ListTags.as_byte()];
+    let reply = dispatch(&router, &req).await;
+    let decoded: Vec<crate::cluster::tags::TagEntry> =
+        serde_json::from_slice(&reply).unwrap();
+    assert_eq!(decoded.len(), 2);
+    assert_eq!(decoded[0].key, "alpha");
+    assert_eq!(decoded[1].key, "bravo");
+    assert_eq!(decoded[0].decode_value().unwrap(), [1u8; 32]);
+}
+
+#[tokio::test]
+async fn end_to_end_pin_lookup_delete_over_real_quic() {
+    // Full flow: publish a tag → look it up → list → delete → confirm gone.
+    let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
+    let (_tmp, router) = router_with_full_stack("a", next_port()).await;
+
+    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 key = "clawverse:main:latest-cache";
+    let value = [0xaau8; 32];
+
+    // Miss first.
+    assert!(call_get_tag(&conn, key).await.unwrap().is_none());
+    // Publish.
+    call_put_tag(&conn, key, &value).await.unwrap();
+    // Hit.
+    assert_eq!(call_get_tag(&conn, key).await.unwrap(), Some(value));
+    // List sees it.
+    let list = call_list_tags(&conn).await.unwrap();
+    assert_eq!(list.len(), 1);
+    assert_eq!(list[0].key, key);
+    // Delete.
+    assert!(call_delete_tag(&conn, key).await.unwrap());
+    // Gone.
+    assert!(call_get_tag(&conn, key).await.unwrap().is_none());
+    assert!(!call_delete_tag(&conn, key).await.unwrap()); // second delete
+    assert!(call_list_tags(&conn).await.unwrap().is_empty());
+
+    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 call_get_chunk_verifies_returned_hash() {
     let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();

claw-store/src/cluster/services.rs

@@ -16,6 +16,7 @@ use crate::cluster::blob::BlobStore;
 use crate::cluster::gossip::ClusterGossip;
 use crate::cluster::refs::RefStore;
 use crate::cluster::rpc::{serve_connection, RpcRouter};
+use crate::cluster::tags::TagStore;
 use crate::cluster::transport::{NodeIdentity, QuicServer};
 use crate::config::ClusterConfig;
 use anyhow::{Context, Result};
@@ -50,6 +51,10 @@ pub struct ClusterServices {
     /// alongside the blob store — a node with one gets the other, so
     /// the entire Phase 5 substrate is enabled by a single config field.
     pub ref_store: Option<Arc<RefStore>>,
+    /// Local tag store (Phase 5d). Backs `PutTag` / `GetTag` /
+    /// `DeleteTag` / `ListTags` — the human-readable pin layer over
+    /// raw refs. Opened alongside the blob store for the same reason.
+    pub tag_store: Option<Arc<TagStore>>,
     /// QUIC accept-loop task. `None` when `[cluster.tls]` was absent
     /// and RPC therefore didn't come up.
     accept_task: Option<JoinHandle<()>>,
@@ -124,6 +129,19 @@ impl ClusterServices {
             None => None,
         };
 
+        // Tag store (Phase 5d) sits next to the ref store. Same
+        // rationale: the whole Phase 5 substrate follows blob_store_root.
+        let tag_store: Option<Arc<TagStore>> = match blob_store_root.as_ref() {
+            Some(root) => {
+                let tags_dir = root.join("tags-db");
+                let store = TagStore::open(tags_dir.clone())
+                    .with_context(|| format!("opening tag store at {}", tags_dir.display()))?;
+                tracing::info!("tag store opened at {}", tags_dir.display());
+                Some(Arc::new(store))
+            }
+            None => None,
+        };
+
         // RPC server + accept loop — only when TLS material is configured.
         let accept_task = match &cluster.tls {
             Some(tls) => {
@@ -147,6 +165,9 @@ impl ClusterServices {
                 if let Some(store) = &ref_store {
                     router = router.with_ref_store(store.clone());
                 }
+                if let Some(store) = &tag_store {
+                    router = router.with_tag_store(store.clone());
+                }
                 let router = Arc::new(router);
                 tracing::info!("cluster RPC server listening on {}", bind);
                 Some(tokio::spawn(async move {
@@ -177,6 +198,7 @@ impl ClusterServices {
             gossip,
             blob_store,
             ref_store,
+            tag_store,
             accept_task,
             metric_task,
         })

claw-store/src/cluster/tags.rs

@@ -0,0 +1,433 @@
+//! Named-tag store (Phase 5d).
+//!
+//! Human-readable string keys mapped to 32-byte values. Sits alongside
+//! the raw [`RefStore`](crate::cluster::refs::RefStore) — refs are for
+//! content-addressed automatic lookup (`fingerprint → BlobId`), tags
+//! are for operator-visible pins (`clawverse:main:latest-cache → BlobId`).
+//!
+//! # Layout
+//!
+//! ```text
+//! <root>/
+//!   tags/<hh>/<blake3_key_hex>.tag  — length-prefixed key + 32-byte value
+//!   .tmp/                            — atomic-rename staging
+//! ```
+//!
+//! On-disk record format:
+//!
+//! ```text
+//! key_len:u16 (LE) | key_bytes | value:32bytes
+//! ```
+//!
+//! The filename is `blake3(key)` (hex) so arbitrary UTF-8 keys land at
+//! deterministic paths without filesystem escaping. Two-char bucket
+//! prefix keeps directory fan-out bounded.
+
+use anyhow::{bail, Context, Result};
+use std::path::{Path, PathBuf};
+use tokio::io::AsyncWriteExt;
+
+/// Longest tag key we accept. 4 KiB is generous; anything longer is
+/// almost certainly a bug on the caller side.
+pub const MAX_TAG_KEY_BYTES: usize = 4096;
+
+/// A tag entry as returned by `list`.
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub struct TagEntry {
+    pub key: String,
+    /// Hex-encoded 32-byte value.
+    pub value_hex: String,
+}
+
+impl TagEntry {
+    /// Parse the 32-byte value out of `value_hex`. Errors if the hex
+    /// is malformed or wrong length.
+    pub fn decode_value(&self) -> Result<[u8; 32]> {
+        if self.value_hex.len() != 64 {
+            bail!(
+                "value_hex length {} (expected 64)",
+                self.value_hex.len()
+            );
+        }
+        let mut out = [0u8; 32];
+        for i in 0..32 {
+            out[i] = u8::from_str_radix(&self.value_hex[i * 2..i * 2 + 2], 16)
+                .with_context(|| format!("parsing byte at hex offset {}", i * 2))?;
+        }
+        Ok(out)
+    }
+}
+
+/// Directory-backed tag store.
+#[derive(Debug, Clone)]
+pub struct TagStore {
+    root: PathBuf,
+}
+
+impl TagStore {
+    /// Open (create if missing) a tag store rooted at `root`. Creates
+    /// `tags/` and `.tmp/` subdirs. Safe on existing stores.
+    pub fn open(root: PathBuf) -> Result<Self> {
+        std::fs::create_dir_all(root.join("tags"))
+            .with_context(|| format!("creating tags dir under {}", root.display()))?;
+        std::fs::create_dir_all(root.join(".tmp"))
+            .with_context(|| format!("creating .tmp dir under {}", root.display()))?;
+        Ok(Self { root })
+    }
+
+    pub fn root(&self) -> &Path {
+        &self.root
+    }
+
+    /// Set a tag → value mapping. Overwrites any existing value.
+    /// Errors when `key` is empty or exceeds [`MAX_TAG_KEY_BYTES`].
+    pub async fn put(&self, key: &str, value: &[u8; 32]) -> Result<()> {
+        validate_key(key)?;
+        let final_path = self.tag_path(key);
+        if let Some(parent) = final_path.parent() {
+            tokio::fs::create_dir_all(parent).await.with_context(|| {
+                format!("creating tag bucket {}", parent.display())
+            })?;
+        }
+        let bytes = encode_record(key, value);
+        self.atomic_write(&final_path, &bytes).await
+    }
+
+    /// Look up a tag. `None` when no tag with that name has been set.
+    pub async fn get(&self, key: &str) -> Result<Option<[u8; 32]>> {
+        validate_key(key)?;
+        let path = self.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, value) = decode_record(&bytes).with_context(|| {
+            format!("decoding tag record at {}", path.display())
+        })?;
+        if parsed_key != key {
+            // Collision on the filename hash: filename derived from
+            // blake3(key) matched but the stored key inside differs.
+            // BLAKE3 collisions are astronomically unlikely; if this
+            // ever fires it's file-system corruption, not the CRDT
+            // model.
+            bail!(
+                "tag record at {} has key {:?} but was requested as {:?}",
+                path.display(),
+                parsed_key,
+                key
+            );
+        }
+        Ok(Some(value))
+    }
+
+    /// Delete a tag. Returns `true` if a tag was removed, `false` if
+    /// no such tag existed.
+    pub async fn delete(&self, key: &str) -> Result<bool> {
+        validate_key(key)?;
+        let path = self.tag_path(key);
+        match tokio::fs::remove_file(&path).await {
+            Ok(()) => Ok(true),
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
+            Err(e) => Err(anyhow::Error::from(e)),
+        }
+    }
+
+    /// Whether a tag with the given name exists.
+    pub async fn contains(&self, key: &str) -> Result<bool> {
+        validate_key(key)?;
+        match tokio::fs::metadata(self.tag_path(key)).await {
+            Ok(_) => Ok(true),
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
+            Err(e) => Err(anyhow::Error::from(e)),
+        }
+    }
+
+    /// List every stored tag. Sorted by key for deterministic output.
+    /// O(N) filesystem walk — fine for our fleet-scale tag counts.
+    pub async fn list(&self) -> Result<Vec<TagEntry>> {
+        let tags_root = self.root.join("tags");
+        let mut entries: Vec<TagEntry> = Vec::new();
+
+        let mut top = tokio::fs::read_dir(&tags_root)
+            .await
+            .with_context(|| format!("reading {}", tags_root.display()))?;
+        while let Some(bucket) = top.next_entry().await? {
+            if !bucket.file_type().await?.is_dir() {
+                continue;
+            }
+            let mut inner = tokio::fs::read_dir(bucket.path()).await?;
+            while let Some(entry) = inner.next_entry().await? {
+                if !entry.file_type().await?.is_file() {
+                    continue;
+                }
+                let bytes = tokio::fs::read(entry.path()).await?;
+                if let Ok((key, value)) = decode_record(&bytes) {
+                    entries.push(TagEntry {
+                        key,
+                        value_hex: hex32(&value),
+                    });
+                }
+            }
+        }
+        entries.sort_by(|a, b| a.key.cmp(&b.key));
+        Ok(entries)
+    }
+
+    fn tag_path(&self, key: &str) -> PathBuf {
+        let hash_hex = hex32(blake3::hash(key.as_bytes()).as_bytes());
+        self.root
+            .join("tags")
+            .join(&hash_hex[..2])
+            .join(format!("{hash_hex}.tag"))
+    }
+
+    async fn atomic_write(&self, final_path: &Path, bytes: &[u8]) -> Result<()> {
+        let tmp_dir = self.root.join(".tmp");
+        let tmp_name = format!(
+            "{}.{}",
+            std::process::id(),
+            RANDOM_SUFFIX.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
+        );
+        let tmp_path = tmp_dir.join(tmp_name);
+        {
+            let mut f = tokio::fs::File::create(&tmp_path).await.with_context(|| {
+                format!("creating tmp tag {}", tmp_path.display())
+            })?;
+            f.write_all(bytes).await?;
+            f.sync_all().await?;
+        }
+        tokio::fs::rename(&tmp_path, final_path)
+            .await
+            .with_context(|| {
+                format!(
+                    "renaming {} → {}",
+                    tmp_path.display(),
+                    final_path.display()
+                )
+            })?;
+        Ok(())
+    }
+}
+
+static RANDOM_SUFFIX: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
+
+/// Sanity checks on a tag key before we touch the filesystem.
+fn validate_key(key: &str) -> Result<()> {
+    if key.is_empty() {
+        bail!("tag key cannot be empty");
+    }
+    if key.len() > MAX_TAG_KEY_BYTES {
+        bail!(
+            "tag key length {} exceeds cap {}",
+            key.len(),
+            MAX_TAG_KEY_BYTES
+        );
+    }
+    Ok(())
+}
+
+/// Encode a `(key, value)` pair as `key_len:u16 (LE) || key_bytes || value`.
+pub fn encode_record(key: &str, value: &[u8; 32]) -> Vec<u8> {
+    let key_bytes = key.as_bytes();
+    let mut out = Vec::with_capacity(2 + key_bytes.len() + 32);
+    let key_len = key_bytes.len() as u16;
+    out.extend_from_slice(&key_len.to_le_bytes());
+    out.extend_from_slice(key_bytes);
+    out.extend_from_slice(value);
+    out
+}
+
+/// Decode `key_len:u16 (LE) || key_bytes || value` back into
+/// `(key, value)`. Errors on truncation or invalid UTF-8.
+pub fn decode_record(bytes: &[u8]) -> Result<(String, [u8; 32])> {
+    if bytes.len() < 2 {
+        bail!("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 + 32;
+    if bytes.len() != expected {
+        bail!(
+            "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("tag key is not valid UTF-8")?
+        .to_string();
+    let mut value = [0u8; 32];
+    value.copy_from_slice(&bytes[2 + key_len..]);
+    Ok((key, value))
+}
+
+fn hex32(bytes: &[u8; 32]) -> String {
+    let mut out = String::with_capacity(64);
+    for b in bytes {
+        out.push_str(&format!("{b:02x}"));
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn open() -> (tempfile::TempDir, TagStore) {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let store = TagStore::open(tmp.path().to_path_buf()).unwrap();
+        (tmp, store)
+    }
+
+    #[tokio::test]
+    async fn open_creates_layout() {
+        let (tmp, store) = open();
+        assert!(tmp.path().join("tags").is_dir());
+        assert!(tmp.path().join(".tmp").is_dir());
+        assert_eq!(store.root(), tmp.path());
+    }
+
+    #[tokio::test]
+    async fn get_returns_none_for_missing() {
+        let (_tmp, store) = open();
+        assert!(store.get("clawverse:main").await.unwrap().is_none());
+        assert!(!store.contains("clawverse:main").await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn put_and_get_round_trip() {
+        let (_tmp, store) = open();
+        let value = [0xabu8; 32];
+        store.put("clawverse:main:latest", &value).await.unwrap();
+        assert_eq!(
+            store.get("clawverse:main:latest").await.unwrap(),
+            Some(value)
+        );
+        assert!(store.contains("clawverse:main:latest").await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn put_overwrites_prior_value() {
+        let (_tmp, store) = open();
+        let v1 = [0x11u8; 32];
+        let v2 = [0x22u8; 32];
+        store.put("key", &v1).await.unwrap();
+        store.put("key", &v2).await.unwrap();
+        assert_eq!(store.get("key").await.unwrap(), Some(v2));
+    }
+
+    #[tokio::test]
+    async fn delete_returns_true_for_existing_and_false_for_missing() {
+        let (_tmp, store) = open();
+        let value = [0u8; 32];
+        store.put("bye", &value).await.unwrap();
+        assert!(store.delete("bye").await.unwrap());
+        assert!(!store.delete("bye").await.unwrap());
+        assert!(store.get("bye").await.unwrap().is_none());
+    }
+
+    #[tokio::test]
+    async fn put_rejects_empty_key() {
+        let (_tmp, store) = open();
+        let err = store.put("", &[0u8; 32]).await.unwrap_err().to_string();
+        assert!(err.contains("cannot be empty"), "unexpected: {err}");
+    }
+
+    #[tokio::test]
+    async fn put_rejects_oversize_key() {
+        let (_tmp, store) = open();
+        let key = "x".repeat(MAX_TAG_KEY_BYTES + 1);
+        let err = store.put(&key, &[0u8; 32]).await.unwrap_err().to_string();
+        assert!(err.contains("exceeds cap"), "unexpected: {err}");
+    }
+
+    #[tokio::test]
+    async fn list_returns_all_tags_sorted() {
+        let (_tmp, store) = open();
+        store.put("bravo", &[0x2u8; 32]).await.unwrap();
+        store.put("alpha", &[0x1u8; 32]).await.unwrap();
+        store.put("charlie", &[0x3u8; 32]).await.unwrap();
+        let list = store.list().await.unwrap();
+        assert_eq!(list.len(), 3);
+        assert_eq!(list[0].key, "alpha");
+        assert_eq!(list[1].key, "bravo");
+        assert_eq!(list[2].key, "charlie");
+    }
+
+    #[tokio::test]
+    async fn list_is_empty_on_fresh_store() {
+        let (_tmp, store) = open();
+        assert!(store.list().await.unwrap().is_empty());
+    }
+
+    #[tokio::test]
+    async fn keys_with_slashes_and_colons_round_trip() {
+        // Real-world tags look like "org/repo:branch:name".
+        let (_tmp, store) = open();
+        let key = "osobh/clawverse:main:latest-cache";
+        let value = [0xffu8; 32];
+        store.put(key, &value).await.unwrap();
+        assert_eq!(store.get(key).await.unwrap(), Some(value));
+        let list = store.list().await.unwrap();
+        assert_eq!(list.len(), 1);
+        assert_eq!(list[0].key, key);
+    }
+
+    #[test]
+    fn encode_and_decode_round_trip() {
+        let value = [0x55u8; 32];
+        let bytes = encode_record("some:key/here", &value);
+        let (key, decoded_value) = decode_record(&bytes).unwrap();
+        assert_eq!(key, "some:key/here");
+        assert_eq!(decoded_value, value);
+    }
+
+    #[test]
+    fn decode_rejects_short_record() {
+        let err = decode_record(&[0]).unwrap_err().to_string();
+        assert!(err.contains("too short"), "unexpected: {err}");
+    }
+
+    #[test]
+    fn decode_rejects_length_mismatch() {
+        // Length prefix says 5 bytes, but the buffer is not
+        // (2 + 5 + 32) = 39 bytes long.
+        let mut bytes = 5u16.to_le_bytes().to_vec();
+        bytes.extend_from_slice(b"hello");
+        // No value bytes appended — truncated.
+        let err = decode_record(&bytes).unwrap_err().to_string();
+        assert!(err.contains("does not match"), "unexpected: {err}");
+    }
+
+    #[test]
+    fn decode_rejects_non_utf8_key() {
+        let value = [0u8; 32];
+        // 3 bytes of key, all invalid UTF-8 leading bytes.
+        let mut bytes = 3u16.to_le_bytes().to_vec();
+        bytes.extend_from_slice(&[0xff, 0xff, 0xff]);
+        bytes.extend_from_slice(&value);
+        let err = decode_record(&bytes).unwrap_err().to_string();
+        assert!(err.contains("UTF-8"), "unexpected: {err}");
+    }
+
+    #[test]
+    fn tag_entry_decode_value_round_trip() {
+        let value = [0x77u8; 32];
+        let entry = TagEntry {
+            key: "k".into(),
+            value_hex: hex32(&value),
+        };
+        assert_eq!(entry.decode_value().unwrap(), value);
+    }
+
+    #[test]
+    fn tag_entry_decode_value_rejects_bad_hex() {
+        let entry = TagEntry {
+            key: "k".into(),
+            value_hex: "z".repeat(64),
+        };
+        assert!(entry.decode_value().is_err());
+    }
+}