Phase 6c: tags as files in claw-fuse

<mount>/tags/<sanitized-name>   ← file, content = current tag's blob

Layers atop the Phase 6a/6b mount. Operator can now `cat` a named
build cache without translating a tag → blob-id first:

  cat <mount>/tags/clawverse:main:latest-cache | tar -tvzf -

Slash → underscore for filenames (tag keys like `a/b:c` land as
`a_b:c`), tags with control chars or NUL are dropped from the
listing. Non-existent value blobs also drop.

Tag file inodes 1_000..9_999. Each getattr / read re-resolves
the tag key (they're mutable — a `pin --replace` under a tag
should show the new blob without unmount).

Reused alloc_blob_ino so tag reads share inodes with /blobs/<hex>
where possible. TagStore opened at `<data_dir>/tags-db` matching
the daemon's convention.

claw-store/src/claw_fuse.rs

@@ -43,6 +43,7 @@ mod zfs;
 
 use cluster::blob::{BlobId, BlobStore};
 use cluster::snapshot::SnapshotStore;
+use cluster::tags::TagStore;
 
 const TTL: Duration = Duration::from_secs(1);
 // Inode number layout:
@@ -54,6 +55,9 @@ const TTL: Duration = Duration::from_secs(1);
 const ROOT_INO: u64 = 1;
 const BLOBS_DIR_INO: u64 = 2;
 const SNAPSHOTS_DIR_INO: u64 = 3;
+const TAGS_DIR_INO: u64 = 4;
+// Tag file inodes: 1_000..9_999.
+const FIRST_TAG_INO: u64 = 1_000;
 // Snapshot dir inodes: 10_000..99_999 (up to 90k snapshots).
 const FIRST_SNAPSHOT_INO: u64 = 10_000;
 // Blob file inodes: 100_000..
@@ -63,6 +67,7 @@ const FIRST_BLOB_INO: u64 = 100_000;
 struct ClawFuse {
     store: BlobStore,
     snapshots: SnapshotStore,
+    tags: TagStore,
     /// Runtime for async store calls. fuser is sync so we
     /// block_on inside each callback.
     runtime: tokio::runtime::Runtime,
@@ -74,15 +79,22 @@ struct ClawFuse {
     snapshot_to_ino: HashMap<String, u64>,
     /// inode → snapshot name.
     ino_to_snapshot: HashMap<u64, String>,
+    /// sanitized-tag-name → allocated inode.
+    tag_to_ino: HashMap<String, u64>,
+    /// inode → (sanitized-name, original-tag-key). Tag files
+    /// read the blob whose id is TagStore::get(original_key).
+    ino_to_tag: HashMap<u64, (String, String)>,
     next_blob_ino: u64,
     next_snapshot_ino: u64,
+    next_tag_ino: u64,
 }
 
 impl ClawFuse {
-    fn new(store: BlobStore, snapshots: SnapshotStore) -> Result<Self> {
+    fn new(store: BlobStore, snapshots: SnapshotStore, tags: TagStore) -> Result<Self> {
         Ok(Self {
             store,
             snapshots,
+            tags,
             runtime: tokio::runtime::Builder::new_current_thread()
                 .enable_all()
                 .build()?,
@@ -90,11 +102,45 @@ impl ClawFuse {
             ino_to_hex: HashMap::new(),
             snapshot_to_ino: HashMap::new(),
             ino_to_snapshot: HashMap::new(),
+            tag_to_ino: HashMap::new(),
+            ino_to_tag: HashMap::new(),
             next_blob_ino: FIRST_BLOB_INO,
             next_snapshot_ino: FIRST_SNAPSHOT_INO,
+            next_tag_ino: FIRST_TAG_INO,
         })
     }
 
+    /// Sanitize a tag key for use as a filename. Replace `/`
+    /// with `_` (tag keys often look like `clawverse:main:cache`
+    /// which is fine, but some contain slashes). Reject anything
+    /// with control chars or NUL — those names would surprise
+    /// tools reading the mount. Returns None when the key is
+    /// unrepresentable.
+    fn sanitize_tag_name(key: &str) -> Option<String> {
+        if key.is_empty()
+            || key.len() > 255
+            || key.chars().any(|c| c.is_control() || c == '\0')
+        {
+            return None;
+        }
+        Some(key.replace('/', "_"))
+    }
+
+    fn alloc_tag_ino(&mut self, sanitized: &str, original: &str) -> u64 {
+        let ino = *self
+            .tag_to_ino
+            .entry(sanitized.to_string())
+            .or_insert_with(|| {
+                let n = self.next_tag_ino;
+                self.next_tag_ino += 1;
+                n
+            });
+        self.ino_to_tag
+            .entry(ino)
+            .or_insert_with(|| (sanitized.to_string(), original.to_string()));
+        ino
+    }
+
     fn alloc_blob_ino(&mut self, hex: &str) -> u64 {
         let ino = *self
             .hex_to_ino
@@ -199,6 +245,52 @@ impl Filesystem for ClawFuse {
             ROOT_INO if name == "snapshots" => {
                 reply.entry(&TTL, &Self::dir_attr(SNAPSHOTS_DIR_INO), 0);
             }
+            ROOT_INO if name == "tags" => {
+                reply.entry(&TTL, &Self::dir_attr(TAGS_DIR_INO), 0);
+            }
+            TAGS_DIR_INO => {
+                // Find a tag whose sanitized name matches `name` AND
+                // whose value blob exists on disk.
+                let entries = match self.runtime.block_on(self.tags.list()) {
+                    Ok(v) => v,
+                    Err(_) => {
+                        reply.error(libc::EIO);
+                        return;
+                    }
+                };
+                let hit = entries.iter().find(|e| {
+                    Self::sanitize_tag_name(&e.key).as_deref() == Some(name)
+                });
+                let hit = match hit {
+                    Some(h) => h,
+                    None => {
+                        reply.error(libc::ENOENT);
+                        return;
+                    }
+                };
+                let value = match hit.decode_value() {
+                    Ok(v) => v,
+                    Err(_) => {
+                        reply.error(libc::EIO);
+                        return;
+                    }
+                };
+                let blob_id = BlobId::from_bytes(value);
+                let size = match self
+                    .runtime
+                    .block_on(self.store.load_manifest(&blob_id))
+                    .ok()
+                    .flatten()
+                {
+                    Some(m) => m.total_size,
+                    None => {
+                        reply.error(libc::ENOENT);
+                        return;
+                    }
+                };
+                let ino = self.alloc_tag_ino(name, &hit.key);
+                reply.entry(&TTL, &Self::file_attr(ino, size), 0);
+            }
             BLOBS_DIR_INO => match self.resolve_hex(name) {
                 Some((ino, size)) => reply.entry(&TTL, &Self::file_attr(ino, size), 0),
                 None => reply.error(libc::ENOENT),
@@ -264,12 +356,38 @@ impl Filesystem for ClawFuse {
 
     fn getattr(&mut self, _req: &Request, ino: u64, _fh: Option<u64>, reply: ReplyAttr) {
         match ino {
-            ROOT_INO | BLOBS_DIR_INO | SNAPSHOTS_DIR_INO => {
+            ROOT_INO | BLOBS_DIR_INO | SNAPSHOTS_DIR_INO | TAGS_DIR_INO => {
                 reply.attr(&TTL, &Self::dir_attr(ino))
             }
             n if (FIRST_SNAPSHOT_INO..FIRST_BLOB_INO).contains(&n) => {
                 reply.attr(&TTL, &Self::dir_attr(ino));
             }
+            n if (FIRST_TAG_INO..FIRST_SNAPSHOT_INO).contains(&n) => {
+                // Tag file: re-resolve size via TagStore lookup.
+                let (_sanitized, orig) = match self.ino_to_tag.get(&n).cloned() {
+                    Some(pair) => pair,
+                    None => {
+                        reply.error(libc::ENOENT);
+                        return;
+                    }
+                };
+                let value = match self.runtime.block_on(self.tags.get(&orig)).ok().flatten() {
+                    Some(v) => v,
+                    None => {
+                        reply.error(libc::ENOENT);
+                        return;
+                    }
+                };
+                let blob_id = BlobId::from_bytes(value);
+                let size = self
+                    .runtime
+                    .block_on(self.store.load_manifest(&blob_id))
+                    .ok()
+                    .flatten()
+                    .map(|m| m.total_size)
+                    .unwrap_or(0);
+                reply.attr(&TTL, &Self::file_attr(ino, size));
+            }
             _ => {
                 let hex = match self.ino_to_hex.get(&ino).cloned() {
                     Some(h) => h,
@@ -312,6 +430,26 @@ impl Filesystem for ClawFuse {
                 entries.push((ROOT_INO, FileType::Directory, "..".into()));
                 entries.push((BLOBS_DIR_INO, FileType::Directory, "blobs".into()));
                 entries.push((SNAPSHOTS_DIR_INO, FileType::Directory, "snapshots".into()));
+                entries.push((TAGS_DIR_INO, FileType::Directory, "tags".into()));
+            }
+            TAGS_DIR_INO => {
+                entries.push((TAGS_DIR_INO, FileType::Directory, ".".into()));
+                entries.push((ROOT_INO, FileType::Directory, "..".into()));
+                let all = match self.runtime.block_on(self.tags.list()) {
+                    Ok(v) => v,
+                    Err(_) => {
+                        reply.error(libc::EIO);
+                        return;
+                    }
+                };
+                for entry in all {
+                    let sanitized = match Self::sanitize_tag_name(&entry.key) {
+                        Some(s) => s,
+                        None => continue,
+                    };
+                    let ino = self.alloc_tag_ino(&sanitized, &entry.key);
+                    entries.push((ino, FileType::RegularFile, sanitized));
+                }
             }
             BLOBS_DIR_INO => {
                 entries.push((BLOBS_DIR_INO, FileType::Directory, ".".into()));
@@ -396,6 +534,11 @@ impl Filesystem for ClawFuse {
         _lock: Option<u64>,
         reply: ReplyData,
     ) {
+        // Two flavors of file:
+        // * blob file (inode ≥ FIRST_BLOB_INO): direct blob read.
+        // * tag file (FIRST_TAG_INO..FIRST_SNAPSHOT_INO): resolve
+        //   the tag's current value → blob, then read.
+        let blob_id = if ino >= FIRST_BLOB_INO {
             let hex = match self.ino_to_hex.get(&ino).cloned() {
                 Some(h) => h,
                 None => {
@@ -403,12 +546,32 @@ impl Filesystem for ClawFuse {
                     return;
                 }
             };
-        let blob_id = match BlobId::from_hex(&hex) {
+            match BlobId::from_hex(&hex) {
                 Ok(id) => id,
                 Err(_) => {
                     reply.error(libc::EIO);
                     return;
                 }
+            }
+        } else if (FIRST_TAG_INO..FIRST_SNAPSHOT_INO).contains(&ino) {
+            let (_sanitized, orig) = match self.ino_to_tag.get(&ino).cloned() {
+                Some(pair) => pair,
+                None => {
+                    reply.error(libc::ENOENT);
+                    return;
+                }
+            };
+            let value = match self.runtime.block_on(self.tags.get(&orig)).ok().flatten() {
+                Some(v) => v,
+                None => {
+                    reply.error(libc::ENOENT);
+                    return;
+                }
+            };
+            BlobId::from_bytes(value)
+        } else {
+            reply.error(libc::ENOENT);
+            return;
         };
         let bytes = match self.runtime.block_on(self.store.get_bytes(&blob_id)) {
             Ok(Some(b)) => b,
@@ -451,7 +614,13 @@ fn main() -> Result<()> {
         .with_context(|| format!("opening blob store at {}", cli.data_dir.display()))?;
     let snapshots = SnapshotStore::open(cli.data_dir.clone())
         .with_context(|| format!("opening snapshot store at {}", cli.data_dir.display()))?;
-    let fs = ClawFuse::new(store, snapshots)?;
+    // TagStore lives under <data_dir>/tags-db by convention (see
+    // cmd_cluster_gc in main.rs). If it's not there yet, TagStore::open
+    // creates it — empty listing is fine.
+    let tag_dir = cli.data_dir.join("tags-db");
+    let tags = TagStore::open(tag_dir.clone())
+        .with_context(|| format!("opening tag store at {}", tag_dir.display()))?;
+    let fs = ClawFuse::new(store, snapshots, tags)?;
 
     let mut opts = vec![
         MountOption::RO,