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,335 @@
+//! 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());
+    }
+
+    /// End-to-end: seed a bare git repo in a tempdir, ensure it into
+    /// a fresh repo_root, verify cached=true on the second call, then
+    /// release. Requires `git` on PATH; marked `#[ignore]` so CI
+    /// without git-installed runners skips it silently.
+    #[tokio::test]
+    #[ignore]
+    async fn ensure_then_cached_then_release() {
+        let tmp = tempfile::tempdir().unwrap();
+        let source = tmp.path().join("source.git");
+        let repo_root = tmp.path().join("repos");
+
+        // Init a bare-ish source repo with one commit on branch `main`.
+        let seed_dir = tmp.path().join("seed");
+        std::fs::create_dir_all(&seed_dir).unwrap();
+        let git = |args: &[&str], cwd: &std::path::Path| {
+            let out = std::process::Command::new("git")
+                .args(args)
+                .current_dir(cwd)
+                .output()
+                .unwrap();
+            assert!(
+                out.status.success(),
+                "git {:?} failed: {}",
+                args,
+                String::from_utf8_lossy(&out.stderr)
+            );
+        };
+        git(&["init", "-b", "main"], &seed_dir);
+        git(&["config", "user.email", "t@t"], &seed_dir);
+        git(&["config", "user.name", "t"], &seed_dir);
+        std::fs::write(seed_dir.join("README"), "hi").unwrap();
+        git(&["add", "."], &seed_dir);
+        git(&["commit", "-m", "seed"], &seed_dir);
+        git(
+            &["clone", "--bare", seed_dir.to_str().unwrap(), source.to_str().unwrap()],
+            tmp.path(),
+        );
+
+        let req = RepoEnsureRequest {
+            url: format!("file://{}", source.display()),
+            git_ref: "main".into(),
+            workspace: "workspace:test".into(),
+        };
+
+        // First ensure → fresh clone.
+        let r1 = ensure_repo(&repo_root, &req).await.unwrap();
+        assert!(!r1.cached, "first ensure should not be cached");
+        assert!(!r1.head_sha.is_empty());
+        assert!(std::path::Path::new(&r1.path).join(".git").exists());
+
+        // Second ensure → cache hit.
+        let r2 = ensure_repo(&repo_root, &req).await.unwrap();
+        assert!(r2.cached, "second ensure should hit cache");
+        assert_eq!(r1.head_sha, r2.head_sha);
+        assert_eq!(r1.path, r2.path);
+
+        // Release removes it.
+        let rel = release_repo(
+            &repo_root,
+            &RepoReleaseRequest {
+                url: req.url.clone(),
+                git_ref: req.git_ref.clone(),
+                workspace: req.workspace.clone(),
+            },
+        )
+        .await
+        .unwrap();
+        assert!(rel.removed);
+        assert!(!std::path::Path::new(&r1.path).exists());
+
+        // Idempotent release.
+        let rel2 = release_repo(
+            &repo_root,
+            &RepoReleaseRequest {
+                url: req.url,
+                git_ref: req.git_ref,
+                workspace: req.workspace,
+            },
+        )
+        .await
+        .unwrap();
+        assert!(!rel2.removed);
+    }
+
+    #[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,43 @@ 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))
+                        }
+                    };
+                let _ = tokio::fs::create_dir_all(&root).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,

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

@@ -192,6 +192,43 @@ pub async fn call_dashboard_storage(conn: &Connection) -> Result<DashboardStorag
     serde_json::from_slice(&reply).context("decoding DashboardStorageReply JSON")
 }
 
+/// Phase 9 R1b: convenience wrapper for [`Method::RepoEnsure`].
+/// Materializes `(url, git_ref)` on the connected peer under the
+/// caller-provided workspace namespace and returns the resulting
+/// on-disk path + head sha. Cached reply = the checkout was already
+/// present with a valid `.git`.
+pub async fn call_repo_ensure(
+    conn: &Connection,
+    req: &crate::cluster::repo_ensure::RepoEnsureRequest,
+) -> Result<crate::cluster::repo_ensure::RepoEnsureReply> {
+    let payload = serde_json::to_vec(req).context("encoding RepoEnsureRequest")?;
+    let reply = rpc_call(conn, Method::RepoEnsure, &payload).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 RepoEnsureReply JSON")
+}
+
+/// Phase 9 R1b: convenience wrapper for [`Method::RepoRelease`].
+/// Removes the on-disk checkout for `(url, git_ref)` under the
+/// caller's workspace. `removed=false` when nothing was on disk to
+/// begin with (still `Ok`).
+pub async fn call_repo_release(
+    conn: &Connection,
+    req: &crate::cluster::repo_ensure::RepoReleaseRequest,
+) -> Result<crate::cluster::repo_ensure::RepoReleaseReply> {
+    let payload = serde_json::to_vec(req).context("encoding RepoReleaseRequest")?;
+    let reply = rpc_call(conn, Method::RepoRelease, &payload).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 RepoReleaseReply 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).

claw-store/src/cluster/services.rs

@@ -226,6 +226,15 @@ impl ClusterServices {
                     r = r.with_tag_store(store.clone());
                 }
                 r = r.with_outbound_client(outbound_client);
+                // Phase 9 R1a wiring: enable RepoEnsure/RepoRelease when
+                // the daemon has a blob_store_root (which is the
+                // canonical anchor for all fleet on-disk state). Repos
+                // materialize under <blob_store_root>/repos/<workspace>/…
+                if let Some(root) = blob_store_root.as_ref() {
+                    let repo_root = root.join("repos");
+                    let _ = std::fs::create_dir_all(&repo_root);
+                    r = r.with_repo_root(repo_root);
+                }
                 let router = Arc::new(r);
 
                 let server =

claw-store/src/daemon.rs

@@ -7,7 +7,7 @@ use crate::manifest::Manifest;
 use crate::snapshot;
 use crate::sync::{SyncQueue, drain_sync_queue};
 use crate::zfs::SystemZfs;
-use anyhow::Result;
+use anyhow::{Context, Result};
 use chrono::Utc;
 use sysinfo::{ProcessRefreshKind, RefreshKind, System};
 use tokio::time::{interval, Duration};
@@ -36,7 +36,14 @@ pub async fn run(cfg: Config, mut manifest: Manifest) -> Result<()> {
             let hot_dir = cfg.hot.path.clone();
             let hot_max_bytes = cfg.hot.max_gb.saturating_mul(1024 * 1024 * 1024);
             let blob_root = cluster_cfg.blob_store_root.clone();
-            match ClusterServices::start(
+            // Fail fast rather than degrade silently: a bind failure here is
+            // almost always a boot-time race against DHCP/network-online
+            // (the bind address isn't assigned to the interface yet). The
+            // systemd unit has `Restart=on-failure`; exiting lets it retry
+            // a few seconds later once the network is actually up, instead
+            // of leaving the daemon running indefinitely with no gossip,
+            // RPC, or Prometheus endpoint and no visible failure state.
+            let svc = ClusterServices::start(
                 cluster_cfg,
                 cfg.node.name.clone(),
                 hot_dir,
@@ -44,21 +51,14 @@ pub async fn run(cfg: Config, mut manifest: Manifest) -> Result<()> {
                 blob_root,
             )
             .await
-            {
+            .context("starting cluster services")?;
-                Ok(svc) => {
+            tracing::info!(
-                    tracing::info!(
+                rpc_enabled = svc.rpc_enabled(),
-                        rpc_enabled = svc.rpc_enabled(),
+                blob_store_enabled = svc.blob_store_enabled(),
-                        blob_store_enabled = svc.blob_store_enabled(),
+                zone = %cluster_cfg.zone,
-                        zone = %cluster_cfg.zone,
+                "cluster services online"
-                        "cluster services online"
+            );
-                    );
+            Some(svc)
-                    Some(svc)
-                }
-                Err(e) => {
-                    tracing::error!(error = %e, "cluster services failed to start; continuing without cluster");
-                    None
-                }
-            }
         }
         None => {
             tracing::info!("no [cluster] section in config; running standalone");

claw-store/src/serve_v2.rs

@@ -91,9 +91,30 @@ impl V2State {
             .join("aggregator-sessions.json");
         let sessions = SessionStore::load(sessions_path)
             .map_err(|e| anyhow::anyhow!("loading session store: {e}"))?;
+        // Bug fix 2026-07-31: the fleet view previously never included
+        // the node actually serving the dashboard — `cluster.peers` is
+        // by definition every *other* node, so hitting a given node's
+        // `/api/v2/fleet` directly silently dropped that node from its
+        // own view (looked like "node X is missing" from the UI, even
+        // though X was perfectly healthy — it just never queried
+        // itself). Fix: synthesize a self `PeerEntry` from our own
+        // gossip bind address and include it in the fan-out list, same
+        // as any other peer. `peer_rpc_addr` derives the RPC port from
+        // `lan_addr`/`tailscale_addr` via the fleet's +1 convention, so
+        // this resolves to the same `bind_rpc_lan`/`bind_rpc_tailscale`
+        // the daemon actually listens on.
+        let self_peer = PeerEntry {
+            name: cfg.node.name.clone(),
+            zone: cluster.zone.clone(),
+            lan_addr: cluster.bind_lan,
+            tailscale_addr: cluster.bind_tailscale,
+        };
+        let mut peers = cluster.peers.clone();
+        peers.push(self_peer);
+
         Ok(Self {
             aggregator_name: cfg.node.name.clone(),
-            peers: cluster.peers.clone(),
+            peers,
             client: std::sync::Arc::new(client),
             default_rpc_port_offset: 1,
             api_token: cfg.api_token.clone(),
@@ -542,6 +563,34 @@ impl AuthedCaller {
             }
         }
     }
+
+    /// Phase 9 R1b: returns the workspace this caller may act on for
+    /// repo-ensure operations. Namespaced callers are pinned to their
+    /// own namespace; admin/open callers must provide one explicitly
+    /// in the request body. Explicit user-supplied workspace is only
+    /// honored for admin/open — a namespaced caller supplying a
+    /// mismatched workspace is a forbidden write.
+    pub fn resolve_workspace(&self, requested: Option<&str>) -> Result<String, (StatusCode, String)> {
+        match self {
+            AuthedCaller::Admin | AuthedCaller::Open => match requested {
+                Some(w) if !w.is_empty() => Ok(w.to_string()),
+                _ => Err((
+                    StatusCode::BAD_REQUEST,
+                    "workspace is required for unnamespaced callers".to_string(),
+                )),
+            },
+            AuthedCaller::Namespaced { namespace } => match requested {
+                None => Ok(namespace.clone()),
+                Some(w) if w == namespace => Ok(namespace.clone()),
+                Some(w) => Err((
+                    StatusCode::FORBIDDEN,
+                    format!(
+                        "workspace \"{w}\" is outside your namespace \"{namespace}\""
+                    ),
+                )),
+            },
+        }
+    }
 }
 
 // ── write-through fan-out (Phase 9 F1) ──────────────────────────
@@ -724,6 +773,217 @@ async fn fanout_delete_tag(s: &V2State, name: &str) -> Vec<PeerResult> {
     out
 }
 
+// ── Phase 9 R1b: repo-ensure fan-out ────────────────────────────
+
+#[derive(Deserialize)]
+pub struct RepoEnsureBody {
+    pub url: String,
+    pub git_ref: String,
+    /// Optional for namespaced tokens (server pins to the caller's
+    /// namespace); required for admin/open callers.
+    #[serde(default)]
+    pub workspace: Option<String>,
+}
+
+#[derive(Serialize)]
+pub struct RepoPeerResult {
+    pub peer: String,
+    pub ok: bool,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub path: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub head_sha: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub cached: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub removed: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub error: Option<String>,
+}
+
+#[derive(Serialize)]
+pub struct RepoFanoutReply {
+    pub url: String,
+    pub git_ref: String,
+    pub workspace: String,
+    pub peers: Vec<RepoPeerResult>,
+    pub all_ok: bool,
+}
+
+async fn handle_repos_ensure(
+    State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
+    Json(body): Json<RepoEnsureBody>,
+) -> Result<Json<RepoFanoutReply>, (StatusCode, String)> {
+    if body.url.trim().is_empty() {
+        return Err((StatusCode::BAD_REQUEST, "url cannot be empty".into()));
+    }
+    if body.git_ref.trim().is_empty() {
+        return Err((StatusCode::BAD_REQUEST, "git_ref cannot be empty".into()));
+    }
+    let workspace = caller.resolve_workspace(body.workspace.as_deref())?;
+    let req = crate::cluster::repo_ensure::RepoEnsureRequest {
+        url: body.url.clone(),
+        git_ref: body.git_ref.clone(),
+        workspace: workspace.clone(),
+    };
+    let results = fanout_repo_ensure(&s, &req).await;
+    let all_ok = results.iter().all(|r| r.ok);
+    Ok(Json(RepoFanoutReply {
+        url: body.url,
+        git_ref: body.git_ref,
+        workspace,
+        peers: results,
+        all_ok,
+    }))
+}
+
+async fn handle_repos_release(
+    State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
+    Json(body): Json<RepoEnsureBody>,
+) -> Result<Json<RepoFanoutReply>, (StatusCode, String)> {
+    if body.url.trim().is_empty() {
+        return Err((StatusCode::BAD_REQUEST, "url cannot be empty".into()));
+    }
+    if body.git_ref.trim().is_empty() {
+        return Err((StatusCode::BAD_REQUEST, "git_ref cannot be empty".into()));
+    }
+    let workspace = caller.resolve_workspace(body.workspace.as_deref())?;
+    let req = crate::cluster::repo_ensure::RepoReleaseRequest {
+        url: body.url.clone(),
+        git_ref: body.git_ref.clone(),
+        workspace: workspace.clone(),
+    };
+    let results = fanout_repo_release(&s, &req).await;
+    let all_ok = results.iter().all(|r| r.ok);
+    Ok(Json(RepoFanoutReply {
+        url: body.url,
+        git_ref: body.git_ref,
+        workspace,
+        peers: results,
+        all_ok,
+    }))
+}
+
+async fn fanout_repo_ensure(
+    s: &V2State,
+    req: &crate::cluster::repo_ensure::RepoEnsureRequest,
+) -> Vec<RepoPeerResult> {
+    let mut set = JoinSet::new();
+    for peer in &s.peers {
+        let peer = peer.clone();
+        let req = req.clone();
+        let s = s.clone_shallow();
+        set.spawn(async move {
+            let peer_name = peer.name.clone();
+            let res = async {
+                let conn = s.dial(&peer).await?;
+                let r = crate::cluster::rpc::call_repo_ensure(&conn, &req).await;
+                conn.close(quinn::VarInt::from_u32(0), b"done");
+                r
+            }
+            .await;
+            match res {
+                Ok(reply) => RepoPeerResult {
+                    peer: peer_name,
+                    ok: true,
+                    path: Some(reply.path),
+                    head_sha: Some(reply.head_sha),
+                    cached: Some(reply.cached),
+                    removed: None,
+                    error: None,
+                },
+                Err(e) => RepoPeerResult {
+                    peer: peer_name,
+                    ok: false,
+                    path: None,
+                    head_sha: None,
+                    cached: None,
+                    removed: None,
+                    error: Some(e.to_string()),
+                },
+            }
+        });
+    }
+    let mut out = Vec::new();
+    while let Some(joined) = set.join_next().await {
+        match joined {
+            Ok(r) => out.push(r),
+            Err(e) => out.push(RepoPeerResult {
+                peer: "<join-error>".into(),
+                ok: false,
+                path: None,
+                head_sha: None,
+                cached: None,
+                removed: None,
+                error: Some(e.to_string()),
+            }),
+        }
+    }
+    out.sort_by(|a, b| a.peer.cmp(&b.peer));
+    out
+}
+
+async fn fanout_repo_release(
+    s: &V2State,
+    req: &crate::cluster::repo_ensure::RepoReleaseRequest,
+) -> Vec<RepoPeerResult> {
+    let mut set = JoinSet::new();
+    for peer in &s.peers {
+        let peer = peer.clone();
+        let req = req.clone();
+        let s = s.clone_shallow();
+        set.spawn(async move {
+            let peer_name = peer.name.clone();
+            let res = async {
+                let conn = s.dial(&peer).await?;
+                let r = crate::cluster::rpc::call_repo_release(&conn, &req).await;
+                conn.close(quinn::VarInt::from_u32(0), b"done");
+                r
+            }
+            .await;
+            match res {
+                Ok(reply) => RepoPeerResult {
+                    peer: peer_name,
+                    ok: true,
+                    path: None,
+                    head_sha: None,
+                    cached: None,
+                    removed: Some(reply.removed),
+                    error: None,
+                },
+                Err(e) => RepoPeerResult {
+                    peer: peer_name,
+                    ok: false,
+                    path: None,
+                    head_sha: None,
+                    cached: None,
+                    removed: None,
+                    error: Some(e.to_string()),
+                },
+            }
+        });
+    }
+    let mut out = Vec::new();
+    while let Some(joined) = set.join_next().await {
+        match joined {
+            Ok(r) => out.push(r),
+            Err(e) => out.push(RepoPeerResult {
+                peer: "<join-error>".into(),
+                ok: false,
+                path: None,
+                head_sha: None,
+                cached: None,
+                removed: None,
+                error: Some(e.to_string()),
+            }),
+        }
+    }
+    out.sort_by(|a, b| a.peer.cmp(&b.peer));
+    out
+}
+
 // ── auth middleware ─────────────────────────────────────────────
 
 /// Resolve the request's bearer against the aggregator's known
@@ -1085,6 +1345,8 @@ pub fn build(state: Arc<V2State>) -> Router {
         .route("/api/v2/sessions/:id/pin", post(handle_session_pin))
         .route("/api/v2/sessions/:id/renew", post(handle_renew_session))
         .route("/api/v2/sessions/:id/commit", post(handle_commit_session))
+        .route("/api/v2/repos/ensure", post(handle_repos_ensure))
+        .route("/api/v2/repos/release", post(handle_repos_release))
         .route_layer(axum::middleware::from_fn_with_state(state.clone(), v2_auth))
         .with_state(state)
 }