claw-store/src/actions.rs

@@ -128,6 +128,7 @@ mod tests {
             peer: None,
             cluster: None,
             api_token: None,
+            aggregator: None,
         }
     }
 

claw-store/src/cluster.rs

@@ -23,6 +23,7 @@ pub mod metrics;
 pub mod prom;
 pub mod ref_tracking;
 pub mod refs;
+pub mod repo_ensure;
 pub mod rpc;
 pub mod services;
 pub mod snapshot;

claw-store/src/cluster/repo_ensure.rs

@@ -0,0 +1,251 @@
+//! Phase 9 R1a: peer-side git materialization.
+//!
+//! Two RPCs give the aggregator (or any authorized client) a way to
+//! ensure a `(url, git_ref)` pair is checked out on this node under a
+//! caller-provided workspace namespace, and to release it later.
+//!
+//! Nothing here fans out to peers — the aggregator layer is
+//! responsible for calling every node. This module is intentionally
+//! narrow: one node, one path derivation, one shallow clone.
+
+use anyhow::{Context, Result};
+use serde::{Deserialize, Serialize};
+use std::path::{Path, PathBuf};
+use std::time::Duration;
+use tokio::process::Command;
+use tokio::time::timeout;
+
+/// Cap on how long a clone can take. Shallow clones of even a large
+/// repo over LAN complete in seconds; anything past 5 minutes is a
+/// stuck network or a pathologically large tree.
+const CLONE_TIMEOUT: Duration = Duration::from_secs(300);
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct RepoEnsureRequest {
+    pub url: String,
+    pub git_ref: String,
+    pub workspace: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct RepoEnsureReply {
+    pub path: String,
+    pub head_sha: String,
+    /// True when the checkout was already present with a valid `.git`
+    /// and no reclone was needed.
+    pub cached: bool,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct RepoReleaseRequest {
+    pub url: String,
+    pub git_ref: String,
+    pub workspace: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct RepoReleaseReply {
+    /// True when the path existed and was removed. False when nothing
+    /// was on disk to begin with (still success from the caller's POV).
+    pub removed: bool,
+}
+
+/// Sanitize an untrusted path component so it can never escape its
+/// parent. Replaces `/`, `\`, `..`, control characters, and leading
+/// dots. The result is always non-empty and safe to `join` under a
+/// known root.
+fn sanitize_component(input: &str) -> String {
+    if input.is_empty() {
+        return "_".to_string();
+    }
+    let mut out = String::with_capacity(input.len());
+    for ch in input.chars() {
+        let mapped = match ch {
+            '/' | '\\' | ':' | '\0' => '_',
+            c if c.is_control() => '_',
+            c => c,
+        };
+        out.push(mapped);
+    }
+    // Reject traversal — after char-mapping we could still have "..".
+    // Collapse any run of dots at either end to `_` prefix.
+    let trimmed = out.trim_matches('.');
+    if trimmed.is_empty() {
+        return "_".to_string();
+    }
+    // Replace embedded ".." segments defensively.
+    trimmed.replace("..", "__")
+}
+
+/// Deterministic on-disk path for a `(workspace, url, git_ref)` triple.
+/// The URL is hashed so we don't leak credentials or full URLs into
+/// directory names; the ref is sanitized so branch names with slashes
+/// (`feature/x`) don't create nested dirs.
+pub fn derive_path(repo_root: &Path, workspace: &str, url: &str, git_ref: &str) -> PathBuf {
+    let ws = sanitize_component(workspace);
+    let ref_slug = sanitize_component(git_ref);
+    let hex = blake3::hash(url.as_bytes()).to_hex();
+    let leaf = format!("{}-{}", &hex.as_str()[..16], ref_slug);
+    repo_root.join(ws).join(leaf)
+}
+
+/// Run `git -C <path> rev-parse HEAD` and return the resulting sha
+/// as a lowercase hex string. `Err` on any failure (including path
+/// not being a git repo).
+async fn read_head_sha(path: &Path) -> Result<String> {
+    let output = Command::new("git")
+        .arg("-C")
+        .arg(path)
+        .args(["rev-parse", "HEAD"])
+        .output()
+        .await
+        .context("spawning `git rev-parse`")?;
+    if !output.status.success() {
+        anyhow::bail!(
+            "git rev-parse failed at {}: {}",
+            path.display(),
+            String::from_utf8_lossy(&output.stderr).trim()
+        );
+    }
+    let sha = String::from_utf8(output.stdout)
+        .context("git rev-parse output was not utf-8")?
+        .trim()
+        .to_string();
+    Ok(sha)
+}
+
+/// Ensure `(url, git_ref)` is materialized under `repo_root`. When a
+/// prior checkout already exists and its `.git` resolves cleanly, we
+/// treat that as a cache hit and return without touching disk. On any
+/// mismatch we remove-and-reclone.
+pub async fn ensure_repo(repo_root: &Path, req: &RepoEnsureRequest) -> Result<RepoEnsureReply> {
+    let path = derive_path(repo_root, &req.workspace, &req.url, &req.git_ref);
+
+    if path.join(".git").exists() {
+        if let Ok(sha) = read_head_sha(&path).await {
+            return Ok(RepoEnsureReply {
+                path: path.display().to_string(),
+                head_sha: sha,
+                cached: true,
+            });
+        }
+        // .git present but rev-parse failed — treat as corrupt and
+        // reclone.
+    }
+
+    // Ensure any partial prior attempt is cleared before we clone.
+    if path.exists() {
+        tokio::fs::remove_dir_all(&path)
+            .await
+            .with_context(|| format!("removing stale checkout at {}", path.display()))?;
+    }
+    if let Some(parent) = path.parent() {
+        tokio::fs::create_dir_all(parent)
+            .await
+            .with_context(|| format!("creating parent {}", parent.display()))?;
+    }
+
+    let clone = Command::new("git")
+        .args(["clone", "--depth", "1", "--branch", &req.git_ref])
+        .arg(&req.url)
+        .arg(&path)
+        .output();
+
+    let output = timeout(CLONE_TIMEOUT, clone)
+        .await
+        .with_context(|| format!("git clone timed out after {:?}", CLONE_TIMEOUT))?
+        .context("spawning `git clone`")?;
+
+    if !output.status.success() {
+        // Best-effort cleanup so we don't leave a half-clone behind.
+        let _ = tokio::fs::remove_dir_all(&path).await;
+        anyhow::bail!(
+            "git clone failed for {}@{}: {}",
+            req.url,
+            req.git_ref,
+            String::from_utf8_lossy(&output.stderr).trim()
+        );
+    }
+
+    let sha = read_head_sha(&path).await?;
+    Ok(RepoEnsureReply {
+        path: path.display().to_string(),
+        head_sha: sha,
+        cached: false,
+    })
+}
+
+/// Remove the on-disk checkout for `(url, git_ref)` under `repo_root`.
+/// A missing path is not an error — the caller's precondition
+/// (nothing at this key) is already satisfied.
+pub async fn release_repo(
+    repo_root: &Path,
+    req: &RepoReleaseRequest,
+) -> Result<RepoReleaseReply> {
+    let path = derive_path(repo_root, &req.workspace, &req.url, &req.git_ref);
+    if !path.exists() {
+        return Ok(RepoReleaseReply { removed: false });
+    }
+    tokio::fs::remove_dir_all(&path)
+        .await
+        .with_context(|| format!("removing checkout at {}", path.display()))?;
+    Ok(RepoReleaseReply { removed: true })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn derive_path_is_deterministic() {
+        let root = PathBuf::from("/tmp/repos");
+        let a = derive_path(&root, "workspace:abc", "https://example.com/x.git", "main");
+        let b = derive_path(&root, "workspace:abc", "https://example.com/x.git", "main");
+        assert_eq!(a, b);
+    }
+
+    #[test]
+    fn derive_path_differs_by_ref() {
+        let root = PathBuf::from("/tmp/repos");
+        let a = derive_path(&root, "ws", "https://example.com/x.git", "main");
+        let b = derive_path(&root, "ws", "https://example.com/x.git", "develop");
+        assert_ne!(a, b);
+    }
+
+    #[test]
+    fn derive_path_differs_by_url() {
+        let root = PathBuf::from("/tmp/repos");
+        let a = derive_path(&root, "ws", "https://example.com/x.git", "main");
+        let b = derive_path(&root, "ws", "https://example.com/y.git", "main");
+        assert_ne!(a, b);
+    }
+
+    #[test]
+    fn derive_path_traversal_safe() {
+        let root = PathBuf::from("/tmp/repos");
+        // Malicious workspace tries to escape the root.
+        let p = derive_path(&root, "../etc", "https://example.com/x.git", "main");
+        assert!(
+            p.starts_with(&root),
+            "sanitized workspace must stay under root: got {}",
+            p.display()
+        );
+        assert!(!p.to_string_lossy().contains(".."));
+    }
+
+    #[test]
+    fn derive_path_slashy_ref_is_flat() {
+        let root = PathBuf::from("/tmp/repos");
+        let p = derive_path(&root, "ws", "https://example.com/x.git", "feature/nested");
+        // Ref becomes part of a single filename, not a nested dir.
+        assert_eq!(p.parent().unwrap().parent().unwrap(), root.as_path());
+    }
+
+    #[test]
+    fn sanitize_component_replaces_traversal() {
+        assert_eq!(sanitize_component(".."), "_");
+        assert_eq!(sanitize_component("../etc"), "_etc");
+        assert_eq!(sanitize_component(""), "_");
+        assert_eq!(sanitize_component("a/b\\c:d"), "a_b_c_d");
+    }
+}

claw-store/src/cluster/rpc.rs

@@ -200,6 +200,23 @@ pub enum Method {
     /// `payload`: empty.
     /// Reply: JSON `DashboardStorageReply`.
     DashboardStorage = 0x1d,
+    /// Phase 9 R1a (2026-07-15): shallow-clone a `(url, git_ref)` on
+    /// this peer under a caller-provided workspace namespace. The
+    /// aggregator fans this out to every fleet node; this per-peer
+    /// method is deliberately narrow — one clone, one path.
+    ///
+    /// `payload`: JSON [`crate::cluster::repo_ensure::RepoEnsureRequest`].
+    /// Reply: JSON [`crate::cluster::repo_ensure::RepoEnsureReply`].
+    /// Requires [`RpcRouter::with_repo_root`]; else
+    /// [`ErrorCode::NotConfigured`].
+    RepoEnsure = 0x1e,
+    /// Phase 9 R1a: inverse of [`Method::RepoEnsure`]. Removes the
+    /// on-disk checkout for `(url, git_ref)` under the caller's
+    /// workspace. Idempotent: absent checkout ⇒ `removed=false`.
+    ///
+    /// `payload`: JSON [`crate::cluster::repo_ensure::RepoReleaseRequest`].
+    /// Reply: JSON [`crate::cluster::repo_ensure::RepoReleaseReply`].
+    RepoRelease = 0x1f,
 }
 
 impl Method {
@@ -236,6 +253,8 @@ impl Method {
             0x1b => Some(Method::GetTagExpiry),
             0x1c => Some(Method::DashboardStatus),
             0x1d => Some(Method::DashboardStorage),
+            0x1e => Some(Method::RepoEnsure),
+            0x1f => Some(Method::RepoRelease),
             _ => None,
         }
     }
@@ -743,6 +762,10 @@ pub struct RpcRouter {
     /// `GetRefLocal` (strict local-only). Set at daemon startup by
     /// `ClusterServices` when a NodeIdentity is available.
     outbound_client: Option<Arc<crate::cluster::transport::QuicClient>>,
+    /// Phase 9 R1a: root directory under which `RepoEnsure` materializes
+    /// checkouts. `None` disables both `RepoEnsure` and `RepoRelease`
+    /// (server returns `NotConfigured`).
+    repo_root: Option<std::path::PathBuf>,
 }
 
 /// Result of dispatching a request: either a real reply (`Ok`) or a
@@ -764,9 +787,17 @@ impl RpcRouter {
             local_name,
             local_zone,
             outbound_client: None,
+            repo_root: None,
         }
     }
 
+    /// Phase 9 R1a: enable `RepoEnsure` / `RepoRelease` by attaching a
+    /// root directory. The directory is created on first use.
+    pub fn with_repo_root(mut self, root: std::path::PathBuf) -> Self {
+        self.repo_root = Some(root);
+        self
+    }
+
     /// Enable ref-forwarding on `GetRef` misses by installing the
     /// outbound QUIC client the router will use to dial peers. See
     /// the field's rustdoc for the semantics.
@@ -1108,6 +1139,45 @@ impl RpcRouter {
                     .context("encoding DashboardStorageReply as JSON")?;
                 Ok(HandlerOutcome::Reply(json))
             }
+            Method::RepoEnsure => {
+                let root = match &self.repo_root {
+                    Some(r) => r.clone(),
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
+                };
+                let req: crate::cluster::repo_ensure::RepoEnsureRequest =
+                    match serde_json::from_slice(payload) {
+                        Ok(r) => r,
+                        Err(_) => {
+                            return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
+                        }
+                    };
+                if let Some(parent) = Some(root.as_path()) {
+                    let _ = tokio::fs::create_dir_all(parent).await;
+                }
+                let reply =
+                    crate::cluster::repo_ensure::ensure_repo(&root, &req).await?;
+                let json = serde_json::to_vec(&reply)
+                    .context("encoding RepoEnsureReply as JSON")?;
+                Ok(HandlerOutcome::Reply(json))
+            }
+            Method::RepoRelease => {
+                let root = match &self.repo_root {
+                    Some(r) => r.clone(),
+                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
+                };
+                let req: crate::cluster::repo_ensure::RepoReleaseRequest =
+                    match serde_json::from_slice(payload) {
+                        Ok(r) => r,
+                        Err(_) => {
+                            return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
+                        }
+                    };
+                let reply =
+                    crate::cluster::repo_ensure::release_repo(&root, &req).await?;
+                let json = serde_json::to_vec(&reply)
+                    .context("encoding RepoReleaseReply as JSON")?;
+                Ok(HandlerOutcome::Reply(json))
+            }
             Method::BlobStat => {
                 let store = match &self.blob_store {
                     Some(s) => s,