claw-store/src/cluster/blob.rs

@@ -583,6 +583,31 @@ impl BlobStore {
     /// `chunks_scanned` is 0 (this function doesn't do a full scan;
     /// call [`gc_orphan_chunks`] separately for that).
     pub async fn evict_to_size_cap(&self, max_bytes: u64) -> Result<GcReport> {
+        // No pins → every manifest is a candidate. Delegate.
+        self.evict_to_size_cap_with_pins(
+            max_bytes,
+            &std::collections::HashSet::new(),
+        )
+        .await
+    }
+
+    /// Phase 4 (2026-07-13): pin-aware LRU eviction. `pinned_blobs`
+    /// is the set of blob IDs the caller considers protected from
+    /// eviction — typically the set of every blob referenced by a
+    /// live tag. Pinned manifests are skipped entirely; their chunks
+    /// stay in the referenced set so shared chunks with evicted
+    /// blobs also survive.
+    ///
+    /// The eviction pass may go under cap earlier than a pin-free
+    /// pass would — pinned blobs count against `max_bytes` but can't
+    /// be evicted to make room, so if the pinned footprint alone
+    /// exceeds the cap, we return without doing anything (the caller
+    /// is expected to raise `blob_max_gb` or drop pins).
+    pub async fn evict_to_size_cap_with_pins(
+        &self,
+        max_bytes: u64,
+        pinned_blobs: &std::collections::HashSet<BlobId>,
+    ) -> Result<GcReport> {
         // 1. Compute per-manifest chunk sets + total live size.
         let manifest_summaries = self.collect_manifest_summaries().await?;
         let mut referenced: std::collections::HashMap<ChunkHash, u32> =
@@ -604,12 +629,17 @@ impl BlobStore {
         let mut bytes_reclaimed = 0u64;
 
         // 2. Sort oldest-first + evict manifests until under cap.
+        //    Skip pinned blobs entirely — their chunks stay in
+        //    `referenced` so any shared chunks also stay put.
         let mut summaries = manifest_summaries;
         summaries.sort_by_key(|s| s.manifest_mtime);
         for summary in summaries {
             if current_size <= max_bytes {
                 break;
             }
+            if pinned_blobs.contains(&summary.blob_id) {
+                continue;
+            }
             self.delete_manifest(&summary.blob_id).await?;
             // For each chunk this manifest used: decrement refcount;
             // if it hits zero, delete the chunk file + free its bytes.
@@ -1152,6 +1182,69 @@ mod tests {
         assert_eq!(report.bytes_reclaimed, 0);
     }
 
+    #[tokio::test]
+    async fn evict_with_pins_protects_pinned_blobs_from_eviction() {
+        // Phase 4: 3 blobs, mtime-ordered a < b < c. Pin the OLDEST
+        // (a) — normal LRU would evict a first. With pins, a survives
+        // and b (the next-oldest) is evicted instead.
+        let (_tmp, store) = open_store();
+
+        let a_bytes = vec![0u8; 100_000];
+        let b_bytes = vec![1u8; 100_000];
+        let c_bytes = vec![2u8; 100_000];
+
+        let id_a = store.put_bytes(&a_bytes).await.unwrap();
+        tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
+        let id_b = store.put_bytes(&b_bytes).await.unwrap();
+        tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
+        let id_c = store.put_bytes(&c_bytes).await.unwrap();
+
+        let mut pinned = std::collections::HashSet::new();
+        pinned.insert(id_a);
+
+        // Cap at ~2 blobs. Without pins, a would be evicted; with
+        // the pin, b goes instead.
+        let report = store
+            .evict_to_size_cap_with_pins(250_000, &pinned)
+            .await
+            .unwrap();
+        assert!(report.chunks_removed >= 1);
+
+        assert!(
+            store.load_manifest(&id_a).await.unwrap().is_some(),
+            "pinned blob a must survive"
+        );
+        assert!(
+            store.load_manifest(&id_b).await.unwrap().is_none(),
+            "next-oldest unpinned b was evicted"
+        );
+        assert!(
+            store.load_manifest(&id_c).await.unwrap().is_some(),
+            "newest c stays"
+        );
+    }
+
+    #[tokio::test]
+    async fn evict_with_pins_stops_when_pinned_footprint_dominates() {
+        // Every blob is pinned → nothing to evict → eviction is a
+        // no-op regardless of `max_bytes`.
+        let (_tmp, store) = open_store();
+        let id_a = store.put_bytes(&vec![9u8; 100_000]).await.unwrap();
+        let id_b = store.put_bytes(&vec![8u8; 100_000]).await.unwrap();
+
+        let mut pinned = std::collections::HashSet::new();
+        pinned.insert(id_a);
+        pinned.insert(id_b);
+
+        let report = store
+            .evict_to_size_cap_with_pins(0, &pinned)
+            .await
+            .unwrap();
+        assert_eq!(report.chunks_removed, 0);
+        assert!(store.load_manifest(&id_a).await.unwrap().is_some());
+        assert!(store.load_manifest(&id_b).await.unwrap().is_some());
+    }
+
     #[tokio::test]
     async fn gc_on_empty_store_reports_zero() {
         let (_tmp, store) = open_store();

claw-store/src/cluster/services.rs

@@ -301,6 +301,7 @@ impl ClusterServices {
         let gc_task = match (&blob_store, cluster.gc_interval_hours) {
             (Some(store), Some(hours)) if hours > 0 => {
                 let store = store.clone();
+                let tag_store_for_gc = tag_store.clone();
                 let interval = Duration::from_secs(hours * 3600);
                 let max_gb = cluster.blob_max_gb;
                 Some(tokio::spawn(async move {
@@ -328,11 +329,30 @@ impl ClusterServices {
                         // that actually bounds growth.
                         if let Some(gb) = max_gb {
                             let cap = gb.saturating_mul(1024 * 1024 * 1024);
-                            match store.evict_to_size_cap(cap).await {
+                            // Phase 4 (2026-07-13): pin-aware eviction.
+                            // Any blob referenced by a tag (stamped or
+                            // legacy) survives; pins act as retention
+                            // markers so operators can `claw-cargo pin`
+                            // a build and know it won't be evicted by
+                            // the size cap.
+                            let pinned = match &tag_store_for_gc {
+                                Some(ts) => {
+                                    let raw = ts.pinned_blob_values().await.unwrap_or_default();
+                                    raw.into_iter()
+                                        .map(crate::cluster::blob::BlobId::from_bytes)
+                                        .collect()
+                                }
+                                None => std::collections::HashSet::new(),
+                            };
+                            match store
+                                .evict_to_size_cap_with_pins(cap, &pinned)
+                                .await
+                            {
                                 Ok(r) if r.chunks_removed > 0 => tracing::info!(
                                     chunks_removed = r.chunks_removed,
                                     bytes_reclaimed = r.bytes_reclaimed,
                                     max_gb = gb,
+                                    pinned_blobs = pinned.len(),
                                     "auto-GC evicted LRU blobs to hit size cap"
                                 ),
                                 Ok(_) => {} // under cap already; keep quiet

claw-store/src/cluster/tags.rs

@@ -268,6 +268,67 @@ impl TagStore {
         Ok(entries)
     }
 
+    /// Phase 4 (2026-07-13): union of every 32-byte value referenced
+    /// by a tag in either the legacy `tags/` or the Phase-3c
+    /// `tags-v2/` namespace. Feeds pin-aware eviction: any blob whose
+    /// id appears in this set is protected from LRU eviction.
+    ///
+    /// Bounded by (tags on disk × 32 bytes); a fleet with 1000 tags
+    /// is well under 100 KiB of memory.
+    pub async fn pinned_blob_values(
+        &self,
+    ) -> Result<std::collections::HashSet<[u8; 32]>> {
+        let mut out = std::collections::HashSet::new();
+        // Unstamped: same walk as `list`, but we skip TagEntry hex
+        // encoding and just push raw 32-byte values.
+        let tags_root = self.root.join("tags");
+        if tags_root.is_dir() {
+            let mut top = tokio::fs::read_dir(&tags_root).await?;
+            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 = match tokio::fs::read(entry.path()).await {
+                        Ok(b) => b,
+                        Err(_) => continue,
+                    };
+                    if let Ok((_, value)) = decode_record(&bytes) {
+                        out.insert(value);
+                    }
+                }
+            }
+        }
+        // Stamped: walk tags-v2/, decode as StampedTag, insert value.
+        let stamped_root = self.root.join("tags-v2");
+        if stamped_root.is_dir() {
+            let mut top = tokio::fs::read_dir(&stamped_root).await?;
+            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 = match tokio::fs::read(entry.path()).await {
+                        Ok(b) => b,
+                        Err(_) => continue,
+                    };
+                    if let Ok((_, stamped)) = decode_stamped_record(&bytes) {
+                        out.insert(stamped.value);
+                    }
+                }
+            }
+        }
+        Ok(out)
+    }
+
     fn tag_path(&self, key: &str) -> PathBuf {
         let hash_hex = hex32(blake3::hash(key.as_bytes()).as_bytes());
         self.root
@@ -657,6 +718,43 @@ mod tests {
         assert_eq!(store.get_stamped(k).await.unwrap(), Some(higher));
     }
 
+    #[tokio::test]
+    async fn pinned_blob_values_unions_both_stores() {
+        // Phase 4 (2026-07-13): pin-set feed for LRU eviction. Any
+        // 32-byte value referenced by ANY tag (legacy or stamped)
+        // must show up in the union. Duplicates dedupe naturally.
+        let (_tmp, store) = open();
+
+        // Legacy tag pointing at value V1.
+        let v1 = [0xA1; 32];
+        store.put("clawverse:main:latest", &v1).await.unwrap();
+
+        // Stamped tag pointing at value V2.
+        let v2 = [0xB2; 32];
+        let s2 = StampedTagValue {
+            value: v2,
+            clock: 5,
+            node: [0; 8],
+        };
+        store
+            .put_stamped("clawverse:pr-42:latest", s2)
+            .await
+            .unwrap();
+
+        // Same value V1 also pinned via a stamped tag → union dedupes.
+        let s3 = StampedTagValue {
+            value: v1,
+            clock: 6,
+            node: [0; 8],
+        };
+        store.put_stamped("mirror:main", s3).await.unwrap();
+
+        let pins = store.pinned_blob_values().await.unwrap();
+        assert!(pins.contains(&v1));
+        assert!(pins.contains(&v2));
+        assert_eq!(pins.len(), 2, "duplicate values dedupe");
+    }
+
     #[tokio::test]
     async fn stamped_and_unstamped_stores_are_independent() {
         // put() writes to tags/, put_stamped() writes to tags-v2/.

claw-store/src/main.rs

@@ -246,7 +246,8 @@ async fn main() -> Result<()> {
 // ── cluster peer-status ───────────────────────────────────────────────────────
 
 async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
-    use cluster::blob::BlobStore;
+    use cluster::blob::{BlobId, BlobStore};
+    use cluster::tags::TagStore;
 
     let root = cfg
         .cluster
@@ -267,14 +268,31 @@ async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
         .await
         .context("gc_orphan_chunks failed")?;
 
-    // Phase 2 (optional): LRU eviction to hit a size cap.
+    // Phase 4 (2026-07-13): gather pin set from tag store so evictions
+    // respect them. Cheap even at fleet scale (one 32-byte value per
+    // tag).
+    let tags_dir = root.join("tags-db");
+    let pinned_blobs = if tags_dir.is_dir() {
+        let ts = TagStore::open(tags_dir.clone())
+            .with_context(|| format!("opening tag store at {}", tags_dir.display()))?;
+        ts.pinned_blob_values()
+            .await
+            .context("collecting pinned tag values")?
+            .into_iter()
+            .map(BlobId::from_bytes)
+            .collect::<std::collections::HashSet<_>>()
+    } else {
+        std::collections::HashSet::new()
+    };
+
+    // Phase 2 (optional): pin-aware LRU eviction to hit a size cap.
     let evict = if let Some(gb) = evict_to_gb {
         let cap = gb.saturating_mul(1024 * 1024 * 1024);
         Some(
             store
-                .evict_to_size_cap(cap)
+                .evict_to_size_cap_with_pins(cap, &pinned_blobs)
                 .await
-                .context("evict_to_size_cap failed")?,
+                .context("evict_to_size_cap_with_pins failed")?,
         )
     } else {
         None
@@ -291,6 +309,7 @@ async fn cmd_cluster_gc(cfg: &Config, evict_to_gb: Option<u64>) -> Result<()> {
         println!("lru eviction (cap {} GiB):", evict_to_gb.unwrap_or(0));
         println!("  chunks removed:  {}", evict.chunks_removed);
         println!("  bytes reclaimed: {}", evict.bytes_reclaimed);
+        println!("  pinned blobs:    {}", pinned_blobs.len());
     }
     println!("total elapsed:     {:?}", elapsed);
     println!("────────────────────────────────────────────────────");