diff --git a/crates/walgit-config/src/lib.rs b/crates/walgit-config/src/lib.rs
index e3cdaf15..6a156eb3 100644
--- a/crates/walgit-config/src/lib.rs
+++ b/crates/walgit-config/src/lib.rs
@@ -1002,13 +1002,9 @@ impl Config {
             if path.is_empty() || path.iter().any(std::string::String::is_empty) {
                 continue;
             }
-            let value: toml::Value = v
-                .parse::<toml::Value>()
-                .unwrap_or(toml::Value::String(v.clone()));
             // Apply into a copy and type-check it alone: a bad/unknown key is
             // dropped (WARN) instead of failing every other override with it.
-            let mut trial = doc.clone();
-            let bad = {
+            let apply = |value: toml::Value| {
                 fn set(
                     cur: &mut toml::Table,
                     path: &[String],
@@ -1028,22 +1024,32 @@ impl Config {
                         .ok_or_else(|| format!("{key} is not a table"))?;
                     set(next, rest, value)
                 }
-                match set(&mut trial, &path, value) {
-                    Err(why) => Some(why),
-                    Ok(()) => trial.clone().try_into::<Config>().err().map(|e| {
-                        e.to_string()
-                            .lines()
-                            .next()
-                            .unwrap_or("invalid")
-                            .to_string()
-                    }),
+                let mut trial = doc.clone();
+                set(&mut trial, &path, value)?;
+                match trial.clone().try_into::<Config>() {
+                    Ok(_) => Ok(trial),
+                    Err(e) => Err(e
+                        .to_string()
+                        .lines()
+                        .next()
+                        .unwrap_or("invalid")
+                        .to_string()),
                 }
             };
-            if let Some(why) = bad {
-                ignored.push((k, why));
-            } else {
-                doc = trial;
-                touched = true;
+            // The TOML reading first, then the text itself: a key that takes text keeps a
+            // value that only looks like a number or boolean (a numeric bucket name, the
+            // `0` a duration documents as "off").
+            let literal = toml::Value::String(v.clone());
+            let applied = match v.parse::<toml::Value>() {
+                Ok(value) => apply(value).or_else(|why| apply(literal).map_err(|_| why)),
+                Err(_) => apply(literal),
+            };
+            match applied {
+                Ok(trial) => {
+                    doc = trial;
+                    touched = true;
+                }
+                Err(why) => ignored.push((k, why)),
             }
         }
         // `[placement]` is a host fact set as a GROUP: any WALGIT__PLACEMENT__* override
@@ -1079,6 +1085,10 @@ impl Config {
     pub fn validate(&self) -> Result<()> {
         self.refs.validate()?;
         self.packs.validate()?;
+        anyhow::ensure!(
+            self.server.max_concurrent_per_repo > 0,
+            "server.max_concurrent_per_repo must be positive"
+        );
         anyhow::ensure!(
             self.packfile_uri.uri_min_bytes.as_u64() > 0,
             "packfile_uri.uri_min_bytes must be positive"
@@ -1354,6 +1364,14 @@ mod tests {
         assert_eq!(back.store.bucket, c.store.bucket);
     }
 
+    #[test]
+    fn zero_per_repo_concurrency_is_rejected() {
+        let mut c = Config::default();
+        c.server.max_concurrent_per_repo = 0;
+        let err = c.validate().unwrap_err().to_string();
+        assert!(err.contains("server.max_concurrent_per_repo must be positive"));
+    }
+
     #[test]
     fn example_configurations_parse_and_validate() {
         Config::parse(include_str!("../../../walgit.example.toml")).unwrap();
@@ -1514,6 +1532,47 @@ mod tests {
         .unwrap();
     }
 
+    /// A value that reads as a TOML number or boolean is still text to a key that takes
+    /// text: a numeric bucket name, or the `0` the duration keys document as "off".
+    #[test]
+    fn env_override_text_that_looks_like_a_toml_scalar_applies() {
+        let mut c = Config::default();
+        let ignored = c
+            .apply_env_report(
+                vec![
+                    ("WALGIT__STORE__BUCKET".to_string(), "20260915".to_string()),
+                    ("WALGIT__MAINTENANCE__HOST".to_string(), "true".to_string()),
+                    (
+                        "WALGIT__MAINTENANCE__FOLLOW_INTERVAL".to_string(),
+                        "0".to_string(),
+                    ),
+                    ("WALGIT__WAL__MAX_BATCH".to_string(), "7".to_string()),
+                ]
+                .into_iter(),
+            )
+            .unwrap();
+        assert!(ignored.is_empty(), "{ignored:?}");
+        assert_eq!(c.store.bucket, "20260915");
+        assert_eq!(c.maintenance.host.as_deref(), Some("true"));
+        assert_eq!(c.maintenance.follow_interval, Duration::ZERO);
+        assert_eq!(c.wal.max_batch, 7, "a TOML value still wins where it fits");
+        // A value that fits neither reading is still ignored, with the TOML reading's reason.
+        let ignored = c
+            .apply_env_report(
+                vec![
+                    ("WALGIT__SERVER__HTTP2".to_string(), "1".to_string()),
+                    ("WALGIT__STORE__BUCKET__NAME".to_string(), "b".to_string()),
+                ]
+                .into_iter(),
+            )
+            .unwrap();
+        assert_eq!(ignored.len(), 2, "{ignored:?}");
+        assert!(ignored[0].1.contains("integer"), "{:?}", ignored[0]);
+        assert!(ignored[1].1.contains("not a table"), "{:?}", ignored[1]);
+        assert!(c.server.http2);
+        assert_eq!(c.store.bucket, "20260915");
+    }
+
     #[test]
     fn removed_bundle_configuration_is_rejected() {
         for input in [
diff --git a/crates/walgit-git/src/pack_segments.rs b/crates/walgit-git/src/pack_segments.rs
index 25dbf72d..d02ef05f 100644
--- a/crates/walgit-git/src/pack_segments.rs
+++ b/crates/walgit-git/src/pack_segments.rs
@@ -637,16 +637,19 @@ fn append_index(
 }
 
 fn sort(input: &Path, output: &Path, work: &Path, resources: Resources) -> Result<(), GitError> {
-    run(
-        Command::new("sort")
-            .args(["--stable", "--unique", "--key=1,1", "--parallel=1"])
-            .arg(format!("--buffer-size={}", resources.sort_memory_bytes))
-            .arg("--temporary-directory")
-            .arg(work),
-        input,
-        output,
-        work,
-    )
+    run(&mut sort_command(work, resources), input, output, work)
+}
+
+/// GNU and BSD `sort` both read a bare `--buffer-size` number as KiB; the `b`
+/// suffix keeps the budget in bytes instead of 1024 times larger.
+fn sort_command(work: &Path, resources: Resources) -> Command {
+    let mut command = Command::new("sort");
+    command
+        .args(["--stable", "--unique", "--key=1,1", "--parallel=1"])
+        .arg(format!("--buffer-size={}b", resources.sort_memory_bytes))
+        .arg("--temporary-directory")
+        .arg(work);
+    command
 }
 
 fn difference(a: &Path, b: &Path, output: &Path, work: &Path) -> Result<(), GitError> {
diff --git a/crates/walgit-git/src/pack_segments_tests.rs b/crates/walgit-git/src/pack_segments_tests.rs
index 5a9b9037..ea9883d5 100644
--- a/crates/walgit-git/src/pack_segments_tests.rs
+++ b/crates/walgit-git/src/pack_segments_tests.rs
@@ -305,6 +305,15 @@ async fn empty_groups_conserve_everything_and_invalid_budgets_fail() {
     );
 }
 
+#[test]
+fn sort_budget_is_passed_in_bytes() {
+    let command = sort_command(Path::new("work"), resources());
+    assert!(
+        command.get_args().any(|a| a == "--buffer-size=1048576b"),
+        "{command:?}"
+    );
+}
+
 #[test]
 fn type_split_keeps_hint_bytes_and_rejects_missing_objects() {
     let root = tempfile::tempdir().unwrap();
diff --git a/crates/walgit-server/src/bridge.rs b/crates/walgit-server/src/bridge.rs
index b0a01903..c51ec2fb 100644
--- a/crates/walgit-server/src/bridge.rs
+++ b/crates/walgit-server/src/bridge.rs
@@ -14,7 +14,8 @@
 //!   `…/manifest.pb` — the commit point itself as a notification. Accepted
 //!   shapes: a GCS Pub/Sub push envelope (`message.attributes.objectId`), an S3
 //!   event notification (`Records[].s3.object.key`), or a plain `{"key": "…"}`
-//!   / `{"repo": "owner/name"}`. A non-2xx here is meant to be redelivered;
+//!   (all bucket keys, `store.prefix` included) / `{"repo": "owner/name"}`
+//!   (whatever the prefix). A non-2xx here is meant to be redelivered;
 //! * the sweep (`events.sweep_interval`): every repo — the backstop. A sweep
 //!   that finds unpublished entries means the notifications are not flowing
 //!   and says so (`events_bridge_sweep_found_total`, warn).
@@ -258,8 +259,9 @@ impl Bridge {
     }
 }
 
-/// The object keys (or repository ids) a notification body names. Store-agnostic.
-fn notified_keys(v: &serde_json::Value) -> Vec<String> {
+/// The bucket object keys a notification body names, `store.prefix` included as the
+/// notifications carry it. Store-agnostic.
+fn notified_keys(v: &serde_json::Value, store_prefix: &str) -> Vec<String> {
     let mut keys = Vec::new();
     // GCS → Pub/Sub push envelope.
     let attrs = &v["message"]["attributes"];
@@ -299,8 +301,9 @@ fn notified_keys(v: &serde_json::Value) -> Vec<String> {
     if let Some(k) = v["key"].as_str() {
         keys.push(k.to_string());
     }
+    // A repository id names no object: use the key its manifest has in this bucket.
     if let Some(r) = v["repo"].as_str() {
-        keys.push(format!("repos/{r}/manifest.pb"));
+        keys.push(format!("{store_prefix}repos/{r}/manifest.pb"));
     }
     keys
 }
@@ -324,7 +327,7 @@ pub async fn http_notify(
     let v: serde_json::Value = serde_json::from_slice(&bytes)
         .map_err(|e| ApiError::BadRequest(format!("notify body: {e}")))?;
     let mut reports = Vec::new();
-    for key in notified_keys(&v) {
+    for key in notified_keys(&v, &bridge.store_prefix) {
         match bridge.object_finalized(&key).await {
             Ok(Some(report)) => reports.push(report),
             Ok(None) => {}
@@ -390,7 +393,25 @@ mod tests {
             .manifest_repo("prefix/repos/acme/monorepo/manifest.pb")
             .unwrap();
         assert_eq!(id.to_string(), "acme/monorepo");
+        // Every body shape names the same repository under a non-empty prefix.
+        for body in [
+            serde_json::json!({"repo": "acme/monorepo"}),
+            serde_json::json!({"key": "prefix/repos/acme/monorepo/manifest.pb"}),
+            serde_json::json!({"message": {"attributes": {
+                "eventType": "OBJECT_FINALIZE",
+                "objectId": "prefix/repos/acme/monorepo/manifest.pb"}}}),
+            serde_json::json!({"Records": [{"eventName": "ObjectCreated:Put",
+                "s3": {"object": {"key": "prefix/repos/acme/monorepo/manifest.pb"}}}]}),
+        ] {
+            let ids: Vec<_> = notified_keys(&body, &b.store_prefix)
+                .iter()
+                .filter_map(|k| b.manifest_repo(k))
+                .map(|id| id.to_string())
+                .collect();
+            assert_eq!(ids, ["acme/monorepo"], "{body}");
+        }
         for other in [
+            "repos/t/r/manifest.pb",
             "prefix/repos/t/r/wal/abc.pack",
             "prefix/repos/t/r/events/cursor.json",
             "walgit-go/repos/t/r/manifest.pb",
diff --git a/crates/walgit-server/src/lib.rs b/crates/walgit-server/src/lib.rs
index 5f6c6154..f8017560 100644
--- a/crates/walgit-server/src/lib.rs
+++ b/crates/walgit-server/src/lib.rs
@@ -66,12 +66,12 @@ use std::future::Future;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
-use axum::Router;
 use axum::body::{Body, to_bytes};
 use axum::extract::ConnectInfo;
 use axum::http::{Method, Request};
 use axum::response::{IntoResponse, Response};
 use axum::routing::get;
+use axum::{Extension, Router};
 use bytes::Bytes;
 use metrics_exporter_prometheus::PrometheusHandle;
 use walgit_store::DynStore;
@@ -89,7 +89,8 @@ pub struct AppState {
     /// HTTP requests in flight (counted until the response body is done); on the watchdog line.
     pub inflight: Arc<middleware::Inflight>,
     pub caches: cache::ServerCaches,
-    pub metrics_handle: Arc<PrometheusHandle>,
+    /// Prometheus recorder handle. Absent when `telemetry.metrics` is disabled.
+    pub metrics_handle: Option<Arc<PrometheusHandle>>,
     /// Read-through LFS upstream client (`upstream.lfs`).
     pub lfs_upstream: lfs_upstream::Upstream,
     /// Startup prewarm state (gates /readyz when configured).
@@ -112,7 +113,11 @@ impl AppState {
     ) -> anyhow::Result<Arc<Self>> {
         let registry = walgit_wal::Registry::new(store.clone(), cfg.clone());
         let bridge = bridge::Bridge::new(&cfg, registry.clone());
-        let metrics_handle = metrics::install()?;
+        let metrics_handle = if cfg.telemetry.metrics {
+            Some(metrics::install()?)
+        } else {
+            None
+        };
         let tls = tls::load(&cfg)?;
         if let Some(t) = &tls {
             tracing::info!(fingerprint = %t.fingerprint, mode = ?cfg.server.tls.mode, "TLS terminated in-process");
@@ -148,7 +153,7 @@ pub fn router(state: Arc<AppState>) -> Router {
         .gzip(true)
         .quality(tower_http::CompressionLevel::Fastest);
     // Nothing with content is public: the SPA shell/assets, installer, credential
-    // helper and metrics all sit behind `web::require_auth` (identity from a
+    // helper and enabled metrics all sit behind `web::require_auth` (identity from a
     // bearer token or a session cookie). The JSON API and git
     // endpoints do their own `require_read`/`require_write`; `/_auth/*` is the
     // login flow itself; `/services/public/*` is the installer a not-yet-signed-in
@@ -162,11 +167,26 @@ pub fn router(state: Arc<AppState>) -> Router {
                 .with_state(())
                 .layer(web_compression.clone()),
         )
-        .route("/metrics", get(metrics::metrics_route))
         .layer(axum::middleware::from_fn_with_state(
             state.clone(),
             web::require_auth,
         ));
+    let metrics = if let Some(metrics_handle) = &state.metrics_handle {
+        Router::new()
+            .route("/metrics", get(metrics::metrics_route))
+            .layer(Extension(metrics_handle.clone()))
+            .layer(axum::middleware::from_fn_with_state(
+                state.clone(),
+                web::require_auth,
+            ))
+    } else {
+        // Reserve the top-level name so the SPA's `/{owner}` route cannot turn a
+        // disabled scrape endpoint into a successful HTML response.
+        Router::new().route(
+            "/metrics",
+            get(|| async { axum::http::StatusCode::NOT_FOUND }),
+        )
+    };
 
     Router::new()
         .merge(
@@ -180,6 +200,7 @@ pub fn router(state: Arc<AppState>) -> Router {
                 .layer(web_compression),
         )
         .merge(gated)
+        .merge(metrics)
         .route("/healthz", get(health::healthz))
         .route("/readyz", get(health::readyz))
         // The SDK is a static artefact with no data in it; it must load from a
diff --git a/crates/walgit-server/src/metrics.rs b/crates/walgit-server/src/metrics.rs
index 264809e6..5cd09d48 100644
--- a/crates/walgit-server/src/metrics.rs
+++ b/crates/walgit-server/src/metrics.rs
@@ -1,14 +1,12 @@
-//! Prometheus metrics exporter. Installs a recorder up front (once per process)
-//! and exposes the rendered scrape via `/metrics`.
+//! Prometheus metrics exporter. When enabled, installs a recorder once per
+//! process and exposes the rendered scrape via `/metrics`.
 
 use std::sync::{Arc, OnceLock};
 
-use axum::extract::State;
+use axum::Extension;
 use axum::response::IntoResponse;
 use metrics_exporter_prometheus::PrometheusHandle;
 
-use crate::AppState;
-
 static HANDLE: OnceLock<Arc<PrometheusHandle>> = OnceLock::new();
 
 /// Install the Prometheus recorder once per process and return a shared handle.
@@ -32,13 +30,15 @@ pub fn install() -> anyhow::Result<Arc<PrometheusHandle>> {
 }
 
 /// `GET /metrics`
-pub async fn metrics_route(State(st): State<Arc<AppState>>) -> impl IntoResponse {
-    let body = st.metrics_handle.render();
+pub async fn metrics_route(
+    Extension(handle): Extension<Arc<PrometheusHandle>>,
+) -> impl IntoResponse {
     (
         [(
             axum::http::header::CONTENT_TYPE,
             "text/plain; version=0.0.4",
         )],
-        body,
+        handle.render(),
     )
+        .into_response()
 }
diff --git a/crates/walgit-server/src/pack_lifecycle.rs b/crates/walgit-server/src/pack_lifecycle.rs
index 31cbe5b6..8ae0dec7 100644
--- a/crates/walgit-server/src/pack_lifecycle.rs
+++ b/crates/walgit-server/src/pack_lifecycle.rs
@@ -495,7 +495,7 @@ pub async fn run(
             _ => 2,
         };
         let global = matches!(work, Work::FirstFreeze | Work::Resegment);
-        let pin = handle.pin_local_objects().await;
+        let pin = handle.pin_local_objects().await?;
         let mut input = maintenance_input::prepare(
             handle.local().path().to_path_buf(),
             root,
@@ -701,7 +701,7 @@ pub async fn run(
     {
         let oid = gix_hash::ObjectId::from_hex(commit.oid.as_bytes())?;
         handle.local().set_segmented_midx_mode(true);
-        let pin = handle.pin_local_objects().await;
+        let pin = handle.pin_local_objects().await?;
         handle
             .local()
             .write_verified_midx(&final_ids, oid, pin)
diff --git a/crates/walgit-server/tests/e2e.rs b/crates/walgit-server/tests/e2e.rs
index e37eea27..7cce50a3 100644
--- a/crates/walgit-server/tests/e2e.rs
+++ b/crates/walgit-server/tests/e2e.rs
@@ -298,6 +298,66 @@ async fn admin_create_list_delete() -> TestResult {
     Ok(())
 }
 
+/// `PUT /{owner}/{repo}` answers from the bucket: 201 once, then 409 — on the
+/// instance holding the handle (warm) and on one that never opened it (cold);
+/// concurrent creates of one name have one 201 and one 409, on one instance
+/// and across two.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn create_answers_409_for_an_existing_repository() -> TestResult {
+    let (a, b) = Server::start_pair().await?;
+    let client = reqwest::Client::new();
+    let put = |s: &Server, repo: &str| client.put(format!("{}/t/{repo}", s.base_url)).send();
+    assert_eq!(put(&a, "r").await?.status(), 201);
+    assert_eq!(put(&a, "r").await?.status(), 409, "warm handle");
+    assert_eq!(put(&b, "r").await?.status(), 409, "cold instance");
+    // Deleted on B while A still holds a handle: the bucket, not A's map, answers.
+    assert_eq!(
+        client
+            .delete(format!("{}/t/r", b.base_url))
+            .send()
+            .await?
+            .status(),
+        204
+    );
+    assert_eq!(
+        put(&a, "r").await?.status(),
+        201,
+        "re-create after a delete elsewhere"
+    );
+
+    for (n, (x, y)) in [(&a, &a), (&a, &b)].into_iter().enumerate() {
+        let repo = format!("race{n}");
+        let (rx, ry) = tokio::join!(put(x, &repo), put(y, &repo));
+        let mut codes = [rx?.status().as_u16(), ry?.status().as_u16()];
+        codes.sort_unstable();
+        assert_eq!(codes, [201, 409], "{repo}");
+    }
+    Ok(())
+}
+
+/// `auto_create_on_push`: a push to a missing name creates it (its internal
+/// create treats "exists" as fine), and a create after that is a 409.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn auto_create_on_push_creates_the_repository() -> TestResult {
+    let server = Server::start_with_tweak(|c| c.server.auto_create_on_push = true).await?;
+    let src = TestRepo::synthetic(1, 1)?;
+    for refspec in ["HEAD:refs/heads/main", "HEAD:refs/heads/second"] {
+        git_in(&src, &["push", &server.repo_url("t", "auto"), refspec])?;
+    }
+    let refs = server.ls_remote("t", "auto").await?;
+    assert!(
+        refs.contains("refs/heads/main") && refs.contains("refs/heads/second"),
+        "{refs}"
+    );
+    let status = reqwest::Client::new()
+        .put(format!("{}/t/auto", server.base_url))
+        .send()
+        .await?
+        .status();
+    assert_eq!(status, 409);
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
 async fn concurrent_clones_and_pushes_with_telemetry() -> TestResult {
     // Regression: tracing spans entered across .await under a multi-threaded
diff --git a/crates/walgit-server/tests/events.rs b/crates/walgit-server/tests/events.rs
index 6864dbeb..cf6110f0 100644
--- a/crates/walgit-server/tests/events.rs
+++ b/crates/walgit-server/tests/events.rs
@@ -206,6 +206,73 @@ async fn bridge_publishes_from_cursor_exactly_once() -> TestResult {
     Ok(())
 }
 
+/// With a bucket prefix (`store.prefix`), every notification shape wakes the bridge: bucket
+/// notifications carry the prefixed object key, and a plain `{"repo": …}` names the repository
+/// whatever the prefix. A key outside this walgit's prefix is acked and ignored.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn notify_wakes_the_bridge_under_a_store_prefix() -> TestResult {
+    use walgit_store::ObjectStoreExt;
+    let (url, captured) = webhook().await;
+    let server = Server::start_with_tweak(|c| {
+        bridge_cfg(&url, Duration::ZERO)(c);
+        c.store.prefix = "tenant-a".into();
+    })
+    .await?;
+    server.put_repo("t", "r").await?;
+    assert!(
+        server
+            .store
+            .exists("tenant-a/repos/t/r/manifest.pb")
+            .await?,
+        "the repository lives under the prefix"
+    );
+    let src = TestRepo::synthetic(1, 1)?;
+    git_in(&src, &["commit", "--allow-empty", "-m", "a"])?;
+    git_in(&src, &["branch", "-M", "main"])?;
+    git_in(
+        &src,
+        &["remote", "add", "origin", &server.repo_url("t", "r")],
+    )?;
+    git_in(&src, &["push", "-u", "origin", "main"])?;
+
+    let client = reqwest::Client::new();
+    let notify = |body: serde_json::Value| {
+        let client = client.clone();
+        let url = format!("{}/_events/notify", server.base_url);
+        async move {
+            let resp = client.post(url).json(&body).send().await?;
+            anyhow::ensure!(resp.status() == 200, "{body}: {}", resp.status());
+            anyhow::Ok(resp.json::<serde_json::Value>().await?)
+        }
+    };
+    // Outside the prefix: not this walgit's object.
+    let report = notify(serde_json::json!({"key": "repos/t/r/manifest.pb"})).await?;
+    assert_eq!(report, serde_json::json!([]));
+    assert!(captured.lock().unwrap().is_empty());
+
+    let mut events = 0;
+    for body in [
+        serde_json::json!({"repo": "t/r"}),
+        gcs_notification("tenant-a/repos/t/r/manifest.pb", "OBJECT_FINALIZE"),
+        serde_json::json!({"Records": [{"eventName": "ObjectCreated:Put", "s3": {"object": {"key": "tenant-a/repos/t/r/manifest.pb"}}}]}),
+        serde_json::json!({"key": "tenant-a/repos/t/r/manifest.pb"}),
+    ] {
+        git_in(&src, &["commit", "--allow-empty", "-m", "next"])?;
+        git_in(&src, &["push"])?;
+        events += 1;
+        let report = notify(body.clone()).await?;
+        assert_eq!(report[0]["repo"], "t/r", "{body}: {report}");
+        assert_eq!(
+            report[0]["emitted"],
+            if events == 1 { 2 } else { 1 },
+            "{body}: {report}"
+        );
+    }
+    let got = wait_for(&captured, 5).await;
+    assert_eq!(got.len(), 5);
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn bridge_sweep_timer_publishes_without_notifications() -> TestResult {
     let (url, captured) = webhook().await;
diff --git a/crates/walgit-server/tests/harness.rs b/crates/walgit-server/tests/harness.rs
index 78d0d6b5..bc1f6c05 100644
--- a/crates/walgit-server/tests/harness.rs
+++ b/crates/walgit-server/tests/harness.rs
@@ -109,7 +109,13 @@ impl Server {
         cfg.server.public_url = Some(base_url.clone());
         cfg.validate().context("config validate")?;
 
-        let dyn_store: DynStore = store.clone();
+        // `store.prefix` scopes every key, as `walgit_store::from_config` does for real backends.
+        let prefix = cfg.store_prefix();
+        let dyn_store: DynStore = if prefix.is_empty() {
+            store.clone()
+        } else {
+            Arc::new(walgit_store::Prefixed::new(store.clone(), prefix))
+        };
         let state = AppState::new(Arc::new(cfg), dyn_store).await?;
 
         let registry = state.registry.clone();
diff --git a/crates/walgit-server/tests/metrics.rs b/crates/walgit-server/tests/metrics.rs
new file mode 100644
index 00000000..c07b55d2
--- /dev/null
+++ b/crates/walgit-server/tests/metrics.rs
@@ -0,0 +1,22 @@
+mod harness;
+
+use axum::http::StatusCode;
+
+use harness::Server;
+
+type TestResult = anyhow::Result<()>;
+
+#[tokio::test]
+async fn metrics_config_controls_scrape_endpoint() -> TestResult {
+    let disabled = Server::start_with_tweak(|cfg| cfg.telemetry.metrics = false).await?;
+    assert!(disabled.state.metrics_handle.is_none());
+    let response = reqwest::get(format!("{}/metrics", disabled.base_url)).await?;
+    assert_eq!(response.status(), StatusCode::NOT_FOUND);
+
+    let enabled = Server::start_with_tweak(|cfg| cfg.telemetry.metrics = true).await?;
+    assert!(enabled.state.metrics_handle.is_some());
+    let response = reqwest::get(format!("{}/metrics", enabled.base_url)).await?;
+    assert_eq!(response.status(), StatusCode::OK);
+
+    Ok(())
+}
diff --git a/crates/walgit-server/tests/sim.rs b/crates/walgit-server/tests/sim.rs
index d7a87d08..a0bdd9ea 100644
--- a/crates/walgit-server/tests/sim.rs
+++ b/crates/walgit-server/tests/sim.rs
@@ -1651,8 +1651,29 @@ async fn healthy_request_round_trip_budgets() -> Result<()> {
         cp.as_of.is_some() && cp.first_state_at.is_some(),
         "provenance times carried without a log read: {cp:?}"
     );
+    // Create of an existing repository: the Create PUT's 412 is the answer, warm handle or
+    // cold, with no GET/HEAD in front of it — 1 request each.
+    let mut create_ops = Vec::new();
+    for i in [0, cold] {
+        let before = c.instances[i].link.stats().ops.load(Ordering::Relaxed);
+        let again = c.instances[i]
+            .registry
+            .create(&c.id, ObjectFormat::Sha1)
+            .await;
+        let ops = c.instances[i].link.stats().ops.load(Ordering::Relaxed) - before;
+        ensure!(
+            matches!(again, Err(WalError::AlreadyExists)),
+            "create of an existing repository must be AlreadyExists, got {:?}",
+            again.map(|_| ())
+        );
+        ensure!(
+            ops == 1,
+            "create of an existing repository used {ops} requests, budget 1"
+        );
+        create_ops.push(ops);
+    }
     eprintln!(
-        "healthy request counts: push={push_ops}, warm_refs={warm_ops}, cold_refs={cold_ops}, checkpoint={cp_ops}"
+        "healthy request counts: push={push_ops}, warm_refs={warm_ops}, cold_refs={cold_ops}, checkpoint={cp_ops}, create_existing={create_ops:?}"
     );
     Ok(())
 }
diff --git a/crates/walgit-store/src/gcs.rs b/crates/walgit-store/src/gcs.rs
index fbe4a62c..4dd3b4f7 100644
--- a/crates/walgit-store/src/gcs.rs
+++ b/crates/walgit-store/src/gcs.rs
@@ -1201,9 +1201,13 @@ fn is_retryable(e: &google_cloud_storage::Error) -> bool {
         );
     }
     if let Some(code) = e.http_status_code() {
-        return matches!(code, 503 | 504 | 429 | 500);
+        // The statuses GCS documents as retryable. The client's own policy
+        // retries all of 5xx; 501 is left out because an unimplemented
+        // request will not start working on a second try. Same set as
+        // `s3::is_transient_status`.
+        return matches!(code, 408 | 429 | 500 | 502 | 503 | 504);
     }
-    e.is_connect() || e.is_io() || e.is_transient_and_before_rpc()
+    e.is_connect() || e.is_io() || e.is_timeout() || e.is_transient_and_before_rpc()
 }
 
 fn map_error(key: &str, e: google_cloud_storage::Error) -> StoreError {
@@ -1384,6 +1388,32 @@ mod tests {
         assert!(is_retryable(&e));
     }
 
+    fn http_error(status: u16) -> google_cloud_storage::Error {
+        google_cloud_gax::error::Error::http(status, http::HeaderMap::new(), bytes::Bytes::new())
+    }
+
+    #[test]
+    fn transient_http_statuses_are_retryable() {
+        for status in [408, 429, 500, 502, 503, 504] {
+            assert!(
+                is_retryable(&http_error(status)),
+                "{status} should be retryable"
+            );
+        }
+        for status in [400, 403, 404, 409, 412, 501] {
+            assert!(
+                !is_retryable(&http_error(status)),
+                "{status} should be permanent"
+            );
+        }
+    }
+
+    #[test]
+    fn is_retryable_timeout() {
+        let e = google_cloud_gax::error::Error::timeout("attempt timed out");
+        assert!(is_retryable(&e));
+    }
+
     #[test]
     fn is_not_retryable_not_found() {
         let e = google_cloud_gax::error::Error::service(
diff --git a/crates/walgit-store/src/s3.rs b/crates/walgit-store/src/s3.rs
index d788cc2e..2138941b 100644
--- a/crates/walgit-store/src/s3.rs
+++ b/crates/walgit-store/src/s3.rs
@@ -8,14 +8,15 @@
 //! ## Version tokens
 //!
 //! S3 `ETags` are used as opaque `Version` strings. Quotes are stripped
-//! consistently on read and never stored. For non-multipart uploads the
-//! `ETag` is the MD5 of the content; for multipart uploads it is a compound
-//! hash. Callers never parse the token — equality comparison suffices.
+//! consistently on read and never stored; conditional headers put them back
+//! ([`entity_tag`]). For non-multipart uploads the `ETag` is the MD5 of the
+//! content; for multipart uploads it is a compound hash. Callers never parse
+//! the token — equality comparison suffices.
 //!
 //! ## Conditional PUT
 //!
-//! `PutMode::Create`    → `If-None-Match: *`  (object must not exist).
-//! `PutMode::Update(v)` → `If-Match: <etag>`  (CAS on current `ETag`).
+//! `PutMode::Create`    → `If-None-Match: *`    (object must not exist).
+//! `PutMode::Update(v)` → `If-Match: "<etag>"`  (CAS on current `ETag`).
 //! On failure the SDK returns a `PreconditionFailed` service error; we fill
 //! `current` via a follow-up HEAD when the SDK doesn't include it.
 //!
@@ -124,10 +125,10 @@ impl S3Store {
         let mut builder = self.client.get_object().bucket(&self.bucket).key(key);
 
         if let Some(v) = &opts.if_none_match {
-            builder = builder.if_none_match(v.as_str());
+            builder = builder.if_none_match(entity_tag(v));
         }
         if let Some(v) = &opts.if_match {
-            builder = builder.if_match(v.as_str());
+            builder = builder.if_match(entity_tag(v));
         }
         if let Some(r) = &opts.range {
             builder = builder.range(Self::range_header(r));
@@ -260,7 +261,7 @@ fn is_transient_code(code: &str) -> bool {
 /// HTTP statuses that mean the same, for services whose error codes we do not
 /// recognise (rustfs and other S3-compatible stores do not all use AWS codes).
 fn is_transient_status(status: u16) -> bool {
-    matches!(status, 429 | 500 | 502 | 503 | 504)
+    matches!(status, 408 | 429 | 500 | 502 | 503 | 504)
 }
 
 /// Whether an SDK failure is worth another attempt at walgit's layer.
@@ -393,7 +394,7 @@ impl ObjectStore for S3Store {
                 builder = builder.if_none_match("*");
             }
             PutMode::Update(v) => {
-                builder = builder.if_match(v.as_str());
+                builder = builder.if_match(entity_tag(v));
             }
         }
 
@@ -977,6 +978,20 @@ impl S3Store {
     }
 }
 
+/// A version token as the entity-tag a conditional header carries. `ETag`s are
+/// stored unquoted, but `If-Match` / `If-None-Match` take quoted entity-tags
+/// (RFC 9110 §8.8.3, §13.1.1), the form every response's `ETag` header has.
+/// AWS S3 and rustfs also accept a bare value; other S3-compatible stores do
+/// not document it (Cloudflare R2 documents neither form).
+fn entity_tag(v: &Version) -> String {
+    let s = v.as_str();
+    if s.len() >= 2 && s.starts_with('"') && s.ends_with('"') {
+        s.to_owned()
+    } else {
+        format!("\"{s}\"")
+    }
+}
+
 fn static_credentials(
     access_key: &str,
     secret_key: &str,
@@ -997,13 +1012,16 @@ fn static_credentials(
 // 7. Multipart: CreateMultipartUpload + UploadPart + CompleteMultipartUpload
 //    supported. No conditional headers on Create/Complete (same as real S3).
 // 8. ETags: quoted, MD5 for single-PUT, compound for multipart. Quotes
-//    stripped consistently in our Version.
+//    stripped consistently in our Version, restored on If-Match/If-None-Match
+//    (rustfs honours both forms).
 // 9. force_path_style: required for rustfs local dev.
 
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    use std::sync::Arc;
+
     use aws_sdk_s3::error::SdkError;
     use aws_sdk_s3::operation::list_objects_v2::ListObjectsV2Error;
     use aws_sdk_s3::operation::put_object::PutObjectError;
@@ -1141,6 +1159,119 @@ mod tests {
         ));
     }
 
+    /// A fake S3 that records each request head and answers a PUT with 200 and
+    /// a GET with 304, both with a quoted `ETag` like real S3.
+    async fn recording_s3() -> (String, Arc<parking_lot::Mutex<Vec<String>>>) {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+        let heads: Arc<parking_lot::Mutex<Vec<String>>> = Arc::default();
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let endpoint = format!("http://{}", listener.local_addr().unwrap());
+        let recorded = heads.clone();
+        tokio::spawn(async move {
+            while let Ok((mut sock, _)) = listener.accept().await {
+                let recorded = recorded.clone();
+                tokio::spawn(async move {
+                    let mut buf = Vec::new();
+                    let mut chunk = [0u8; 8192];
+                    let end = loop {
+                        let n = sock.read(&mut chunk).await.unwrap_or(0);
+                        if n == 0 {
+                            return;
+                        }
+                        buf.extend_from_slice(&chunk[..n]);
+                        if let Some(i) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
+                            break i + 4;
+                        }
+                    };
+                    let head = String::from_utf8_lossy(&buf[..end]).into_owned();
+                    let body_len = head
+                        .lines()
+                        .filter_map(|l| l.split_once(':'))
+                        .find(|(k, _)| k.eq_ignore_ascii_case("content-length"))
+                        .and_then(|(_, v)| v.trim().parse::<usize>().ok())
+                        .unwrap_or(0);
+                    while buf.len() < end + body_len {
+                        let n = sock.read(&mut chunk).await.unwrap_or(0);
+                        if n == 0 {
+                            break;
+                        }
+                        buf.extend_from_slice(&chunk[..n]);
+                    }
+                    let resp = if head.starts_with("GET ") {
+                        "HTTP/1.1 304 Not Modified\r\nETag: \"abc\"\r\nConnection: close\r\n\r\n"
+                    } else {
+                        "HTTP/1.1 200 OK\r\nETag: \"def\"\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
+                    };
+                    recorded.lock().push(head);
+                    let _ = sock.write_all(resp.as_bytes()).await;
+                });
+            }
+        });
+        (endpoint, heads)
+    }
+
+    fn header<'a>(head: &'a str, name: &str) -> Option<&'a str> {
+        head.lines()
+            .filter_map(|l| l.split_once(':'))
+            .find(|(k, _)| k.eq_ignore_ascii_case(name))
+            .map(|(_, v)| v.trim())
+    }
+
+    #[test]
+    fn entity_tags_are_quoted_once() {
+        assert_eq!(entity_tag(&Version::new("abc")), "\"abc\"");
+        assert_eq!(entity_tag(&Version::new("\"abc\"")), "\"abc\"");
+        assert_eq!(entity_tag(&Version::new("abc-3")), "\"abc-3\"");
+    }
+
+    /// Versions hold `ETag`s unquoted; on the wire a conditional header carries
+    /// the quoted entity-tag (RFC 9110), the form every `ETag` response has.
+    #[tokio::test]
+    async fn conditional_headers_carry_quoted_entity_tags() {
+        let (endpoint, heads) = recording_s3().await;
+        let store = S3Store {
+            client: client_for(&endpoint),
+            bucket: "b".into(),
+            http: reqwest::Client::new(),
+            multipart_threshold: u64::MAX,
+            multipart_part_size: 8 << 20,
+        };
+        let put = store
+            .put(
+                "manifest.pb",
+                PutBody::Bytes(Bytes::from_static(b"x")),
+                PutMode::Update(Version::new("abc")).into(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(put.version.as_str(), "def", "stored unquoted");
+        let got = store
+            .get(
+                "manifest.pb",
+                GetOptions {
+                    if_none_match: Some(Version::new("abc")),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+        assert!(matches!(got, GetResult::NotModified { .. }));
+        let _ = store
+            .get(
+                "manifest.pb",
+                GetOptions {
+                    if_match: Some(Version::new("abc")),
+                    ..Default::default()
+                },
+            )
+            .await;
+        let heads = heads.lock().clone();
+        assert_eq!(heads.len(), 3, "{heads:#?}");
+        assert_eq!(header(&heads[0], "if-match"), Some("\"abc\""));
+        assert_eq!(header(&heads[1], "if-none-match"), Some("\"abc\""));
+        assert_eq!(header(&heads[2], "if-match"), Some("\"abc\""));
+    }
+
     #[test]
     fn transient_codes_are_recognised() {
         for code in [
@@ -1172,10 +1303,10 @@ mod tests {
 
     #[test]
     fn transient_statuses_are_recognised() {
-        for status in [429, 500, 502, 503, 504] {
+        for status in [408, 429, 500, 502, 503, 504] {
             assert!(is_transient_status(status), "{status} should be transient");
         }
-        for status in [400, 403, 404, 409, 412] {
+        for status in [400, 403, 404, 409, 412, 501] {
             assert!(!is_transient_status(status), "{status} should be permanent");
         }
     }
diff --git a/crates/walgit-wal/src/checkpoint.rs b/crates/walgit-wal/src/checkpoint.rs
index 5834c207..62354a0c 100644
--- a/crates/walgit-wal/src/checkpoint.rs
+++ b/crates/walgit-wal/src/checkpoint.rs
@@ -259,7 +259,7 @@ async fn write_checkpoint_inner(handle: &RepoHandle) -> Result<CheckpointRef, Wa
                         state.packs_revision = updated.revision;
                     }
                 }
-                crate::state::save_state(handle.local.path(), &handle.state.lock().clone())?;
+                handle.persist_state()?;
                 return Ok(cp_ref);
             }
             Err(StoreError::PreconditionFailed { .. }) => {
diff --git a/crates/walgit-wal/src/handle.rs b/crates/walgit-wal/src/handle.rs
index 611fa3b4..89c817f0 100644
--- a/crates/walgit-wal/src/handle.rs
+++ b/crates/walgit-wal/src/handle.rs
@@ -3,7 +3,7 @@
 use std::collections::HashMap;
 use std::sync::{
     Arc,
-    atomic::{AtomicUsize, Ordering},
+    atomic::{AtomicBool, AtomicUsize, Ordering},
 };
 use std::time::Instant;
 
@@ -54,6 +54,9 @@ pub struct RepoHandle {
     /// stuck behind a multi-GB materialization (only removals take the write
     /// lock, briefly).
     pub(crate) pack_mutex: TokioMutex<()>,
+    /// Set when `Registry::create` replaces this repository's local copy: the
+    /// path now belongs to the next generation, so nothing here may touch it.
+    retired: AtomicBool,
 
     // Current manifest (last known). Short critical sections, no await.
     pub(crate) manifest: PLRwLock<Arc<Manifest>>,
@@ -152,6 +155,7 @@ impl RepoHandle {
             rw: Arc::new(TokioRwLock::new(())),
             sync_mutex: TokioMutex::new(()),
             pack_mutex: TokioMutex::new(()),
+            retired: AtomicBool::new(false),
             manifest: PLRwLock::new(Arc::new(manifest)),
             manifest_version: PLMutex::new(version),
             state: PLMutex::new(state),
@@ -356,6 +360,35 @@ impl RepoHandle {
         &self.local
     }
 
+    /// Fails once the local copy was handed to a newer generation of this repository.
+    pub(crate) fn ensure_live(&self) -> Result<(), WalError> {
+        if self.retired.load(Ordering::Acquire) {
+            return Err(WalError::NotFound);
+        }
+        Ok(())
+    }
+
+    pub(crate) fn persist_state(&self) -> Result<(), WalError> {
+        self.ensure_live()?;
+        crate::state::save_state(self.local.path(), &self.state.lock().clone())
+    }
+
+    /// Retire the handle, then wait out every holder that began before it. Each lock is
+    /// taken and released in turn, never held while waiting for the next, so the drain adds
+    /// no lock-order edge (a queued `rw` writer behind a holder of `pack_mutex`/`sync_mutex`
+    /// that wants `rw.read` would deadlock). Every acquirer checks `ensure_live` after it
+    /// holds its lock, so anything that starts later bails before touching the directory.
+    pub(crate) async fn retire_and_drain(&self) {
+        self.retire();
+        drop(self.pack_mutex.lock().await);
+        drop(self.sync_mutex.lock().await);
+        drop(self.rw.write().await);
+    }
+
+    pub(crate) fn retire(&self) {
+        self.retired.store(true, Ordering::Release);
+    }
+
     pub fn store(&self) -> &Prefixed {
         &self.store
     }
@@ -423,8 +456,12 @@ impl RepoHandle {
     /// Pin installed object files across an owned blocking maintenance task.
     /// Unlike a borrowed request guard, cancellation of its caller cannot release
     /// this pin while a detached blocking worker still copies the input files.
-    pub async fn pin_local_objects(&self) -> tokio::sync::OwnedRwLockReadGuard<()> {
-        self.rw.clone().read_owned().await
+    pub async fn pin_local_objects(
+        &self,
+    ) -> Result<tokio::sync::OwnedRwLockReadGuard<()>, WalError> {
+        let pin = self.rw.clone().read_owned().await;
+        self.ensure_live()?;
+        Ok(pin)
     }
 
     pub async fn sync_full(&self) -> Result<crate::sync::ReadGuard<'_>, WalError> {
@@ -512,6 +549,7 @@ impl RepoHandle {
                     arc.pack_mutex.lock(),
                 )
                 .await;
+                arc.ensure_live()?;
                 let manifest = arc.manifest();
                 let task = match arc.begin_task("history-pack", HashMap::new()) {
                     Begin::Started(t) => Some(t),
@@ -595,6 +633,7 @@ impl RepoHandle {
             changed = false,
             entries_applied = 0u64,
         );
+        self.ensure_live()?;
         self.touch();
         // Phase 1 — refs (manifest freshness + ref state), under sync_mutex
         // only; sub-second. Phase 2 — packs, under pack_mutex only. Neither
@@ -619,6 +658,7 @@ impl RepoHandle {
             self.rw.read().instrument(span.clone()),
         )
         .await;
+        self.ensure_live()?;
         Ok(crate::sync::ReadGuard {
             _guard: read_guard,
             handle: self,
@@ -628,6 +668,7 @@ impl RepoHandle {
     /// Manifest freshness check + ref state apply (never packs, never
     /// `rw.write()`: ref files and the gix handle are replaced atomically).
     async fn sync_refs_phase(&self, span: &tracing::Span) -> Result<(), WalError> {
+        self.ensure_live()?;
         if self.freshness_ttl_active() {
             return Ok(());
         }
@@ -639,6 +680,7 @@ impl RepoHandle {
             self.sync_mutex.lock(),
         )
         .await;
+        self.ensure_live()?;
         if self.freshness_ttl_active() {
             return Ok(());
         }
@@ -666,6 +708,7 @@ impl RepoHandle {
             self.pack_mutex.lock(),
         )
         .await;
+        self.ensure_live()?;
         if self.level_satisfied(level) {
             return Ok(());
         }
@@ -824,6 +867,7 @@ impl RepoHandle {
     /// Freshness check + refs apply, with `sync_mutex` and the write lock held
     /// by the caller. Packs are never touched here (see `sync_packs_phase`).
     pub(crate) async fn sync_locked_inner(&self, span: &tracing::Span) -> Result<(), WalError> {
+        self.ensure_live()?;
         let known = self.manifest_version.lock().clone();
         let outcome = crate::sync::freshness_check(&self.store, known.as_ref()).await?;
         match outcome {
@@ -931,6 +975,7 @@ impl RepoHandle {
             self.pack_mutex.lock(),
         )
         .await;
+        self.ensure_live()?;
 
         // Read manifest fresh
         let Some((meta, manifest)) =
diff --git a/crates/walgit-wal/src/publish.rs b/crates/walgit-wal/src/publish.rs
index d6d7db4e..222802f8 100644
--- a/crates/walgit-wal/src/publish.rs
+++ b/crates/walgit-wal/src/publish.rs
@@ -582,6 +582,9 @@ async fn process_batch(handle: &RepoHandle, batch: Vec<PublishRequest>) -> Resul
         // lock that applies refs. Release it before any store or bulk work.
         let (manifest, known_version, mut working_refs) = {
             let _sync = handle.sync_mutex.lock().await;
+            if let Err(e) = handle.ensure_live() {
+                return finish_all_errors(batch, e);
+            }
             let (manifest, version) = handle.manifest_pair();
             let refs = match handle.local.ref_view() {
                 Ok(v) => v,
@@ -865,7 +868,6 @@ async fn process_batch(handle: &RepoHandle, batch: Vec<PublishRequest>) -> Resul
             // every waiter is answered `ok`. A failed apply leaves the version unadvertised, so the next
             // sync sees a change and replays the entry — the copy repairs itself. (Answering an error
             // here produced a durable push that git reported as failed — "0 winners", commit fetchable.)
-            let mut local_ok = true;
             {
                 let _sync_guard = crate::lockwait::timed(
                     "sync_mutex",
@@ -875,6 +877,8 @@ async fn process_batch(handle: &RepoHandle, batch: Vec<PublishRequest>) -> Resul
                     handle.sync_mutex.lock(),
                 )
                 .await;
+                let live = handle.ensure_live().is_ok();
+                let mut local_ok = live;
                 let applied_seq = handle.state.lock().applied_seq;
                 let already_applied = applied_seq >= last_seq;
                 if applied_seq < committed.head_seq
@@ -884,11 +888,9 @@ async fn process_batch(handle: &RepoHandle, batch: Vec<PublishRequest>) -> Resul
                     tracing::warn!(error = %e, "landed push catch-up failed; next sync repairs it");
                     local_ok = false;
                 }
-                for (req, _) in batch
-                    .iter()
-                    .zip(&verified)
-                    .filter(|(_, v)| v.valid && !already_applied && committed.head_seq == last_seq)
-                {
+                for (req, _) in batch.iter().zip(&verified).filter(|(_, v)| {
+                    live && v.valid && !already_applied && committed.head_seq == last_seq
+                }) {
                     if let Err(e) = handle.local.apply_ref_txn(&req.txn, false) {
                         tracing::warn!(repo = %handle.id, seq = last_seq, error = %e, "published (CAS ok), but applying the ref txn to the local copy failed; the next sync replays it");
                         metrics::counter!("walgit_publish_local_apply_failed_total").increment(1);
@@ -921,9 +923,7 @@ async fn process_batch(handle: &RepoHandle, batch: Vec<PublishRequest>) -> Resul
                             state.packs_revision = committed.revision;
                         }
                     }
-                    if let Err(e) =
-                        crate::state::save_state(handle.local.path(), &handle.state.lock().clone())
-                    {
+                    if let Err(e) = handle.persist_state() {
                         tracing::warn!(repo = %handle.id, error = %e, "published (CAS ok), but saving local state failed; the next sync repairs it");
                     }
                 } else if !local_ok {
@@ -1320,7 +1320,7 @@ pub(crate) async fn publish_compact_classified(
                     state.packs_revision = committed.revision;
                 }
             }
-            crate::state::save_state(handle.local.path(), &handle.state.lock().clone())?;
+            handle.persist_state()?;
             sweep_burned(&handle.store, &slot).await;
             return Ok(seq);
         }
@@ -1428,7 +1428,7 @@ pub(crate) async fn annotate_pack_impl(
                     state.manifest_version = Some(meta.version.as_str().to_string());
                     state.revision = updated.revision;
                 }
-                crate::state::save_state(handle.local.path(), &handle.state.lock().clone())?;
+                handle.persist_state()?;
                 return Ok(pack_ref);
             }
             Err(StoreError::PreconditionFailed { .. }) => {
@@ -1466,25 +1466,28 @@ pub(crate) async fn add_pack_impl(
     let checksum = gix_hash::ObjectId::from_hex(hex.as_bytes())
         .map_err(|e| WalError::Corrupt(format!("bad pack name {name}: {e}")))?;
     let dest = handle.local.pack_path(&checksum);
-    std::fs::create_dir_all(
-        dest.parent()
-            .ok_or_else(|| WalError::Corrupt("destination has no parent".into()))?,
-    )?;
-    for (src, dst) in [(pack, dest.clone()), (idx, dest.with_extension("idx"))] {
-        if !dst.exists() && std::fs::hard_link(src, &dst).is_err() {
-            std::fs::copy(src, &dst)?;
+    let info = {
+        let _pin = handle.pin_local_objects().await?;
+        std::fs::create_dir_all(
+            dest.parent()
+                .ok_or_else(|| WalError::Corrupt("destination has no parent".into()))?,
+        )?;
+        for (src, dst) in [(pack, dest.clone()), (idx, dest.with_extension("idx"))] {
+            if !dst.exists() && std::fs::hard_link(src, &dst).is_err() {
+                std::fs::copy(src, &dst)?;
+            }
         }
-    }
-    if let Some(base) = &history_of {
-        handle.local.mark_history_pack(&checksum, base).await?;
-    }
-    handle.local.refresh_async().await?;
-    let info = handle
-        .local
-        .packs()?
-        .into_iter()
-        .find(|p| p.checksum == checksum)
-        .ok_or_else(|| WalError::Corrupt(format!("pack {hex} not visible after install")))?;
+        if let Some(base) = &history_of {
+            handle.local.mark_history_pack(&checksum, base).await?;
+        }
+        handle.local.refresh_async().await?;
+        handle
+            .local
+            .packs()?
+            .into_iter()
+            .find(|p| p.checksum == checksum)
+            .ok_or_else(|| WalError::Corrupt(format!("pack {hex} not visible after install")))?
+    };
     publish_compact_impl(handle, info, Vec::new(), tier).await
 }
 
@@ -1590,7 +1593,7 @@ pub(crate) async fn publish_settings_impl(
                         state.packs_revision = updated.revision;
                     }
                 }
-                crate::state::save_state(handle.local.path(), &handle.state.lock().clone())?;
+                handle.persist_state()?;
                 *handle.effective.lock() = None;
                 sweep_burned(&handle.store, &slot).await;
                 tracing::info!(repo = %handle.id, seq, revision, author, "settings published");
diff --git a/crates/walgit-wal/src/registry.rs b/crates/walgit-wal/src/registry.rs
index 9fc0147e..2066d1fa 100644
--- a/crates/walgit-wal/src/registry.rs
+++ b/crates/walgit-wal/src/registry.rs
@@ -19,6 +19,8 @@ use crate::handle::RepoHandle;
 use crate::state::{RepoState, load_state, save_state};
 use crate::store_proto::get_message;
 
+const RESET_MARKER: &str = "walgit-reset-pending";
+
 pub struct Registry {
     store: DynStore,
     cfg: Arc<walgit_config::Config>,
@@ -83,12 +85,12 @@ impl Registry {
         // Every request that touches a repo goes through here: tag the
         // enclosing `http.request` span (no-op outside one).
         tracing::Span::current().record("repo", tracing::field::display(id));
-        if let Some(h) = self.repos.get(id) {
+        if let Some(h) = self.repos.get(id).filter(|h| h.ensure_live().is_ok()) {
             return Ok(h.clone());
         }
         let gate = self.opening.entry(id.clone()).or_default().clone();
         let _g = gate.lock().await;
-        if let Some(h) = self.repos.get(id) {
+        if let Some(h) = self.repos.get(id).filter(|h| h.ensure_live().is_ok()) {
             return Ok(h.clone());
         }
 
@@ -104,11 +106,20 @@ impl Registry {
         crate::validate_manifest(&manifest)?;
 
         // Open or init local repo (LocalRepo joins owner/name.git onto the root).
-        let local = if let Some(l) = LocalRepo::open(&self.cache_root, id)? {
+        let reset_pending = id.local_dir(&self.cache_root).join(RESET_MARKER).exists();
+        let local = if let Some(l) =
+            LocalRepo::open(&self.cache_root, id)?.filter(|_| !reset_pending)
+        {
             l
         } else {
+            // A directory without HEAD, or still marked by a create, is an unfinished init
+            // or an interrupted reset: whatever it still holds is not this repository's.
+            let cache_root = self.cache_root.clone();
+            let owned_id = id.clone();
             let format = parse_object_format(&manifest.object_format);
-            LocalRepo::init(&self.cache_root, id, format)?
+            tokio::task::spawn_blocking(move || reset_local_copy(&cache_root, &owned_id, format))
+                .await
+                .map_err(|e| WalError::Corrupt(format!("local copy task: {e}")))??
         };
 
         // Load state
@@ -156,6 +167,10 @@ impl Registry {
     /// handle and the local copy. Err(NotFound) if the manifest does not exist.
     /// Other instances notice on their next freshness check (manifest GET -> 404).
     pub async fn delete(&self, id: &RepoId) -> Result<(), WalError> {
+        // Held to the end: an `open` racing the drain would cache a handle on a
+        // manifest and directory that are about to disappear.
+        let gate = self.opening.entry(id.clone()).or_default().clone();
+        let _g = gate.lock().await;
         let prefixed = Prefixed::new(self.store.clone(), id.store_prefix());
         if get_message::<Manifest>(&prefixed, keys::MANIFEST)
             .await?
@@ -163,10 +178,13 @@ impl Registry {
         {
             return Err(WalError::NotFound);
         }
-        // Drop the handle first so no request on this instance publishes into a
-        // prefix that is being removed; in-flight requests hold their own Arc.
-        self.repos.remove(id);
+        // Drop and retire the handle first so no request on this instance publishes
+        // into a prefix that is being removed or touches a later generation's directory.
+        let stale = self.repos.remove(id).map(|(_, handle)| handle);
         self.invalidate_listing();
+        if let Some(handle) = &stale {
+            handle.retire_and_drain().await;
+        }
         // Manifest first: it is the linearization point, so the repo disappears
         // atomically for readers; remaining objects are unreferenced garbage.
         prefixed.delete(keys::MANIFEST, None).await?;
@@ -194,19 +212,16 @@ impl Registry {
     }
 
     /// CAS-create manifest.pb (`PutMode::Create`). Err(AlreadyExists) on 412.
+    /// The Create PUT is the only authority, warm handle or not: a cached handle
+    /// is not evidence the manifest still exists (another host may have deleted
+    /// it) and must not turn a second create into a success.
     pub async fn create(
         &self,
         id: &RepoId,
         format: ObjectFormat,
     ) -> Result<Arc<RepoHandle>, WalError> {
-        if let Some(h) = self.repos.get(id) {
-            return Ok(h.clone());
-        }
         let gate = self.opening.entry(id.clone()).or_default().clone();
         let _g = gate.lock().await;
-        if let Some(h) = self.repos.get(id) {
-            return Ok(h.clone());
-        }
 
         let prefix = id.store_prefix();
         let prefixed = Prefixed::new(self.store.clone(), prefix);
@@ -228,6 +243,24 @@ impl Registry {
             retired_packs: Vec::new(),
         };
 
+        // Marked before the Create, so a reset that fails or never runs after the Create
+        // lands (crash, lost reply) leaves `open` to rebuild instead of serving the old copy.
+        let marker = id.local_dir(&self.cache_root).join(RESET_MARKER);
+        // A marker left by an earlier attempt stays: that Create may have landed.
+        let marked = marker.clone();
+        let marked_here = tokio::task::spawn_blocking(move || {
+            if marked.exists() {
+                return Ok(false);
+            }
+            match std::fs::write(&marked, b"") {
+                Ok(()) => Ok(true),
+                Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
+                Err(e) => Err(WalError::Io(e)),
+            }
+        })
+        .await
+        .map_err(|e| WalError::Corrupt(format!("local copy task: {e}")))??;
+
         let buf = manifest.encode_to_vec();
         match prefixed
             .put(
@@ -238,8 +271,24 @@ impl Registry {
             .await
         {
             Ok(meta) => {
-                // Init local repo
-                let local = LocalRepo::init(&self.cache_root, id, format)?;
+                // The Create landed, so nothing cached or on disk belongs to this
+                // manifest. Evict the stale handle so `open` cannot hand it out, then
+                // wait for its in-flight work before the directory is reset in place:
+                // holders keep the path, so the handle is retired and drained first.
+                // Nothing is cached until the new copy is fully built, so a failure
+                // below leaves `open` to rebuild from the manifest.
+                let stale = self.repos.remove(id).map(|(_, handle)| handle);
+                self.invalidate_listing();
+                if let Some(handle) = &stale {
+                    handle.retire_and_drain().await;
+                }
+                let cache_root = self.cache_root.clone();
+                let owned_id = id.clone();
+                let local = tokio::task::spawn_blocking(move || {
+                    reset_local_copy(&cache_root, &owned_id, format)
+                })
+                .await
+                .map_err(|e| WalError::Corrupt(format!("local copy task: {e}")))??;
 
                 let state = RepoState {
                     manifest_version: Some(meta.version.as_str().to_string()),
@@ -267,12 +316,26 @@ impl Registry {
                 self.invalidate_listing();
                 Ok(handle)
             }
-            Err(StoreError::PreconditionFailed { .. }) => Err(WalError::AlreadyExists),
-            Err(e) => Err(WalError::Store(e)),
+            Err(StoreError::PreconditionFailed { .. }) => {
+                if marked_here {
+                    let _ = tokio::fs::remove_file(&marker).await;
+                }
+                Err(WalError::AlreadyExists)
+            }
+            Err(e) => {
+                // The Create may have landed behind a lost reply: the marker stays, and
+                // no cached handle may keep serving the copy the marker condemns.
+                if let Some((_, stale)) = self.repos.remove(id) {
+                    self.invalidate_listing();
+                    stale.retire_and_drain().await;
+                }
+                Err(WalError::Store(e))
+            }
         }
     }
 
-    /// Open or create.
+    /// Open or create. Losing a create race (another request or host created it
+    /// between our 404 and our Create) is success: open what the winner made.
     pub async fn open_or_create(
         &self,
         id: &RepoId,
@@ -280,7 +343,10 @@ impl Registry {
     ) -> Result<Arc<RepoHandle>, WalError> {
         match self.open(id).await {
             Ok(h) => Ok(h),
-            Err(WalError::NotFound) => self.create(id, format).await,
+            Err(WalError::NotFound) => match self.create(id, format).await {
+                Err(WalError::AlreadyExists) => self.open(id).await,
+                other => other,
+            },
             Err(e) => Err(e),
         }
     }
@@ -440,14 +506,24 @@ impl Registry {
             // excludes leaked/long-lived request ReadGuards. The old code only
             // probed sync_mutex and immediately dropped it, so it could delete
             // packs underneath an active reader.
+            // The opening gate keeps `open` from rebuilding on this directory, or
+            // caching a handle on it, until it is gone.
+            let gate = self.opening.entry(id.clone()).or_default().clone();
+            let Ok(_opening) = gate.try_lock() else {
+                continue;
+            };
             let Ok(_sync) = handle.sync_mutex.try_lock() else {
                 continue;
             };
             let Ok(_write) = handle.rw.try_write() else {
                 continue;
             };
+            if handle.ensure_live().is_err() {
+                continue;
+            }
             let path = handle.local.path().to_path_buf();
             let bytes = dir_size(&path);
+            handle.retire();
             self.repos.remove(id);
             let _ = std::fs::remove_dir_all(&path);
             total_bytes = total_bytes.saturating_sub(bytes);
@@ -461,6 +537,23 @@ impl Registry {
     }
 }
 
+/// HEAD goes first: without it `LocalRepo::open` rejects the directory, so a wipe that
+/// stops midway is rebuilt by the next `open` instead of served as a repository.
+fn reset_local_copy(
+    cache_root: &std::path::Path,
+    id: &RepoId,
+    format: ObjectFormat,
+) -> Result<LocalRepo, WalError> {
+    let dir = id.local_dir(cache_root);
+    let ignore_missing = |result: std::io::Result<()>| match result {
+        Err(e) if e.kind() != std::io::ErrorKind::NotFound => Err(WalError::Io(e)),
+        _ => Ok(()),
+    };
+    ignore_missing(std::fs::remove_file(dir.join("HEAD")))?;
+    ignore_missing(std::fs::remove_dir_all(&dir))?;
+    Ok(LocalRepo::init(cache_root, id, format)?)
+}
+
 fn parse_object_format(s: &str) -> ObjectFormat {
     match s {
         "sha256" => ObjectFormat::Sha256,
@@ -514,3 +607,71 @@ fn disk_usage(path: &std::path::Path) -> Option<(u64, u64)> {
     let avail = st.f_bavail as u64 * st.f_frsize as u64;
     Some((total.saturating_sub(avail), total))
 }
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use walgit_store::memory::MemoryStore;
+
+    use super::*;
+
+    /// A holder of `pack_mutex` that then wants `rw.read` (segmented MIDX
+    /// reconciliation) must not be stuck behind the drain's `rw` writer.
+    #[tokio::test]
+    async fn retire_and_drain_does_not_queue_a_writer_behind_a_pack_mutex_holder() {
+        let cache = tempfile::tempdir().unwrap();
+        let mut cfg = walgit_config::Config::default();
+        cfg.cache.dir = cache.path().to_path_buf();
+        let registry = Registry::new(MemoryStore::shared(), Arc::new(cfg));
+        let id = RepoId::new("test", "drain").unwrap();
+        let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap();
+
+        let pack = handle.pack_mutex.lock().await;
+        let drain = tokio::spawn({
+            let handle = handle.clone();
+            async move { handle.retire_and_drain().await }
+        });
+        while handle.ensure_live().is_ok() {
+            tokio::task::yield_now().await;
+        }
+        tokio::time::sleep(Duration::from_millis(50)).await;
+
+        let pin = tokio::time::timeout(Duration::from_secs(5), handle.pin_local_objects())
+            .await
+            .expect("rw.read must not wait behind the drain");
+        assert!(matches!(pin, Err(WalError::NotFound)));
+
+        drop(pack);
+        tokio::time::timeout(Duration::from_secs(5), drain)
+            .await
+            .expect("drain completes once the holder is gone")
+            .unwrap();
+    }
+
+    /// An `open` in progress owns the directory it is building on; eviction must
+    /// neither remove it nor drop the handle that `open` is about to cache.
+    #[tokio::test]
+    async fn eviction_skips_a_repository_an_open_is_building() {
+        let cache = tempfile::tempdir().unwrap();
+        let mut cfg = walgit_config::Config::default();
+        cfg.cache.dir = cache.path().to_path_buf();
+        cfg.cache.max_bytes = walgit_config::ByteSize::b(0);
+        cfg.cache.evict_idle_after = Duration::ZERO;
+        let registry = Registry::new(MemoryStore::shared(), Arc::new(cfg));
+        let id = RepoId::new("test", "building").unwrap();
+        let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap();
+        tokio::time::sleep(Duration::from_millis(5)).await;
+
+        let gate = registry.opening.entry(id.clone()).or_default().clone();
+        let opening = gate.lock().await;
+        let report = registry.evict_idle().await.unwrap();
+        assert_eq!(report.evicted, 0);
+        assert!(handle.ensure_live().is_ok());
+        assert!(handle.local().path().join("HEAD").exists());
+
+        drop(opening);
+        assert_eq!(registry.evict_idle().await.unwrap().evicted, 1);
+        assert!(handle.ensure_live().is_err());
+    }
+}
diff --git a/crates/walgit-wal/src/snapshots.rs b/crates/walgit-wal/src/snapshots.rs
index f0b04903..5eec4f9b 100644
--- a/crates/walgit-wal/src/snapshots.rs
+++ b/crates/walgit-wal/src/snapshots.rs
@@ -154,6 +154,7 @@ pub(crate) async fn checkpoint_snapshot(
 impl RepoHandle {
     pub async fn fetch_view(&self) -> Result<FetchView, WalError> {
         let _sync = self.sync_mutex.lock().await;
+        self.ensure_live()?;
         let manifest = self.manifest();
         let local = self.local.clone();
         let refs = tokio::task::spawn_blocking(move || local.ref_view())
diff --git a/crates/walgit-wal/src/sync.rs b/crates/walgit-wal/src/sync.rs
index d2711057..c2ad5f4d 100644
--- a/crates/walgit-wal/src/sync.rs
+++ b/crates/walgit-wal/src/sync.rs
@@ -473,6 +473,7 @@ pub(crate) async fn apply_delta(
     new_manifest: &Manifest,
     new_version: &Version,
 ) -> Result<(), WalError> {
+    handle.ensure_live()?;
     let local = &handle.local;
     let current_state = handle.state.lock().clone();
 
@@ -517,7 +518,7 @@ pub(crate) async fn apply_delta(
             state.packs_revision = new_manifest.revision;
         }
     }
-    crate::state::save_state(local.path(), &handle.state.lock().clone())?;
+    handle.persist_state()?;
     local.refresh_async().await?;
     Ok(())
 }
@@ -829,7 +830,7 @@ pub(crate) async fn reconcile_packs_inner(
                         .map_err(|e| WalError::Corrupt(format!("pack checksum: {e}")))
                 })
                 .collect::<Result<Vec<_>, _>>()?;
-            let pin = handle.pin_local_objects().await;
+            let pin = handle.pin_local_objects().await?;
             let repo = local.clone();
             let (refs, pin) = tokio::task::spawn_blocking(move || {
                 let refs = repo.refs();
@@ -861,7 +862,7 @@ pub(crate) async fn reconcile_packs_inner(
         let mut state = handle.state.lock();
         state.packs_revision = manifest.revision;
     }
-    crate::state::save_state(local.path(), &handle.state.lock().clone())?;
+    handle.persist_state()?;
     Ok(())
 }
 
diff --git a/crates/walgit-wal/tests/wal.rs b/crates/walgit-wal/tests/wal.rs
index 8bd97b05..6271b276 100644
--- a/crates/walgit-wal/tests/wal.rs
+++ b/crates/walgit-wal/tests/wal.rs
@@ -22,7 +22,7 @@ use walgit_git::{IngestOptions, ObjectFormat, RepoId};
 use walgit_proto::v1::{EntryKind, RefTransaction, RefUpdate};
 use walgit_store::ObjectStore;
 use walgit_store::memory::MemoryStore;
-use walgit_wal::Registry;
+use walgit_wal::{Registry, WalError};
 
 use std::io::Write;
 
@@ -316,6 +316,472 @@ async fn test_two_registries_cross_sync() {
     );
 }
 
+/// A second create is `AlreadyExists` whether this registry holds the handle
+/// (warm) or never saw the repository (cold): the Create PUT decides, not the map.
+#[tokio::test]
+async fn create_of_an_existing_repository_is_already_exists_warm_or_cold() {
+    let cache = tempfile::tempdir().unwrap();
+    let store = MemoryStore::shared();
+    let warm = Registry::new(
+        store.clone(),
+        Arc::new(make_config(&cache.path().join("a"), 0)),
+    );
+    let cold = Registry::new(
+        store.clone(),
+        Arc::new(make_config(&cache.path().join("b"), 0)),
+    );
+    let id = repo_id("test", "twice");
+
+    warm.create(&id, ObjectFormat::Sha1).await.unwrap();
+    for registry in [&warm, &cold] {
+        assert!(matches!(
+            registry.create(&id, ObjectFormat::Sha1).await,
+            Err(walgit_wal::WalError::AlreadyExists)
+        ));
+    }
+    // The losing creates changed nothing: both still open the one repository.
+    assert_eq!(warm.open(&id).await.unwrap().manifest().revision, 1);
+    assert_eq!(cold.open(&id).await.unwrap().manifest().revision, 1);
+}
+
+/// A create that lands proves nothing cached or on disk belongs to the new
+/// manifest: the copy left by a repository another host deleted must not leak
+/// into it. The old handle is retired, its in-flight work finishes before the
+/// directory is reset, and anything it starts afterwards fails instead of
+/// writing into the new repository.
+#[tokio::test]
+async fn an_open_racing_a_delete_sees_the_repository_gone() {
+    let cache = tempfile::tempdir().unwrap();
+    let store = MemoryStore::shared();
+    let registry = Registry::new(
+        store.clone(),
+        Arc::new(make_config(&cache.path().join("a"), 0)),
+    );
+    let id = repo_id("test", "racing");
+    let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap();
+
+    let in_flight = handle.sync_refs().await.unwrap();
+    let delete = tokio::spawn({
+        let registry = registry.clone();
+        let id = id.clone();
+        async move { registry.delete(&id).await }
+    });
+    loop {
+        match handle.sync_refs().await {
+            Err(WalError::NotFound) => break,
+            Ok(guard) => drop(guard),
+            Err(e) => panic!("unexpected {e}"),
+        }
+        tokio::task::yield_now().await;
+    }
+    let open = tokio::spawn({
+        let registry = registry.clone();
+        let id = id.clone();
+        async move { registry.open(&id).await.map(|_| ()) }
+    });
+    tokio::task::yield_now().await;
+    drop(in_flight);
+
+    delete.await.unwrap().unwrap();
+    assert!(matches!(open.await.unwrap(), Err(WalError::NotFound)));
+    assert!(matches!(registry.open(&id).await, Err(WalError::NotFound)));
+}
+
+#[tokio::test]
+async fn open_rebuilds_a_copy_whose_create_landed_without_its_reset() {
+    let cache = tempfile::tempdir().unwrap();
+    let store = MemoryStore::shared();
+    let cfg = Arc::new(make_config(&cache.path().join("a"), 0));
+    let id = repo_id("test", "crashed");
+
+    let registry = Registry::new(store.clone(), cfg.clone());
+    let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap();
+    let work = WorkRepo::new();
+    let c1 = work.commit("init", "data");
+    let ingested = ingest_pack_data(&handle, work.create_pack()).await.unwrap();
+    handle
+        .publish_push(
+            Some(ingested),
+            make_txn(vec![("refs/heads/main", "", &c1)]),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+    let local_dir = handle.local().path().to_path_buf();
+    drop(handle);
+    drop(registry);
+
+    // Another host deletes and recreates; this host's own Create would have marked the
+    // copy and then died before resetting it.
+    let other = tempfile::tempdir().unwrap();
+    let foreign = Registry::new(store.clone(), Arc::new(make_config(other.path(), 0)));
+    foreign.delete(&id).await.unwrap();
+    foreign.create(&id, ObjectFormat::Sha1).await.unwrap();
+    std::fs::write(local_dir.join("walgit-reset-pending"), b"").unwrap();
+
+    let reopened = Registry::new(store.clone(), cfg).open(&id).await.unwrap();
+    reopened.sync_refs().await.unwrap();
+    assert!(reopened.local().refs().unwrap().refs.is_empty());
+    assert!(!local_dir.join("walgit-reset-pending").exists());
+}
+
+#[cfg(unix)]
+#[tokio::test]
+async fn an_interrupted_rebuild_is_rebuilt_again() {
+    use std::os::unix::fs::PermissionsExt;
+
+    let cache = tempfile::tempdir().unwrap();
+    let store = MemoryStore::shared();
+    let cfg = Arc::new(make_config(&cache.path().join("a"), 0));
+    let id = repo_id("test", "rewiped");
+
+    let registry = Registry::new(store.clone(), cfg.clone());
+    let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap();
+    let local_dir = handle.local().path().to_path_buf();
+    drop(handle);
+    drop(registry);
+
+    std::fs::write(local_dir.join("walgit-reset-pending"), b"").unwrap();
+    let stuck = local_dir.join("stuck");
+    std::fs::create_dir(&stuck).unwrap();
+    std::fs::write(stuck.join("file"), b"").unwrap();
+    std::fs::set_permissions(&stuck, std::fs::Permissions::from_mode(0o555)).unwrap();
+    if std::fs::write(stuck.join("probe"), b"").is_ok() {
+        // Root ignores directory permissions, so the interrupted wipe cannot be staged.
+        std::fs::set_permissions(&stuck, std::fs::Permissions::from_mode(0o755)).unwrap();
+        return;
+    }
+
+    let failed = Registry::new(store.clone(), cfg.clone()).open(&id).await;
+    std::fs::set_permissions(&stuck, std::fs::Permissions::from_mode(0o755)).unwrap();
+    assert!(failed.is_err());
+    assert!(!local_dir.join("HEAD").exists());
+
+    let reopened = Registry::new(store.clone(), cfg).open(&id).await.unwrap();
+    assert!(reopened.local().path().join("HEAD").exists());
+    assert!(!local_dir.join("walgit-reset-pending").exists());
+    assert!(!stuck.exists());
+}
+
+#[tokio::test]
+async fn a_refused_create_keeps_a_marker_an_earlier_create_left() {
+    let cache = tempfile::tempdir().unwrap();
+    let store = MemoryStore::shared();
+    let cfg = Arc::new(make_config(&cache.path().join("a"), 0));
+    let id = repo_id("test", "remarked");
+
+    let registry = Registry::new(store.clone(), cfg.clone());
+    let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap();
+    let work = WorkRepo::new();
+    let c1 = work.commit("init", "data");
+    let ingested = ingest_pack_data(&handle, work.create_pack()).await.unwrap();
+    handle
+        .publish_push(
+            Some(ingested),
+            make_txn(vec![("refs/heads/main", "", &c1)]),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+    let local_dir = handle.local().path().to_path_buf();
+    let marker = local_dir.join("walgit-reset-pending");
+    std::fs::write(&marker, b"").unwrap();
+
+    assert!(matches!(
+        registry.create(&id, ObjectFormat::Sha1).await,
+        Err(WalError::AlreadyExists)
+    ));
+    assert!(marker.exists());
+    drop(handle);
+    drop(registry);
+
+    let reopened = Registry::new(store.clone(), cfg).open(&id).await.unwrap();
+    assert!(!marker.exists());
+    reopened.sync_refs().await.unwrap();
+    assert_eq!(reopened.local().refs().unwrap().refs.len(), 1);
+}
+
+#[tokio::test]
+async fn a_refused_create_keeps_the_existing_local_copy() {
+    let cache = tempfile::tempdir().unwrap();
+    let store = MemoryStore::shared();
+    let cfg = Arc::new(make_config(&cache.path().join("a"), 0));
+    let id = repo_id("test", "kept");
+
+    let registry = Registry::new(store.clone(), cfg.clone());
+    let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap();
+    let work = WorkRepo::new();
+    let c1 = work.commit("init", "data");
+    let ingested = ingest_pack_data(&handle, work.create_pack()).await.unwrap();
+    handle
+        .publish_push(
+            Some(ingested),
+            make_txn(vec![("refs/heads/main", "", &c1)]),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+    assert!(matches!(
+        registry.create(&id, ObjectFormat::Sha1).await,
+        Err(WalError::AlreadyExists)
+    ));
+    let local_dir = handle.local().path().to_path_buf();
+    assert!(!local_dir.join("walgit-reset-pending").exists());
+    drop(handle);
+    drop(registry);
+
+    let reopened = Registry::new(store.clone(), cfg).open(&id).await.unwrap();
+    assert_eq!(reopened.local().refs().unwrap().refs.len(), 1);
+}
+
+#[tokio::test]
+async fn create_after_a_foreign_delete_starts_from_an_empty_local_copy() {
+    let cache = tempfile::tempdir().unwrap();
+    let store = MemoryStore::shared();
+    let registry = Registry::new(
+        store.clone(),
+        Arc::new(make_config(&cache.path().join("a"), 0)),
+    );
+    let id = repo_id("test", "recreated");
+
+    let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap();
+    let work = WorkRepo::new();
+    let c1 = work.commit("init", "data");
+    let ingested = ingest_pack_data(&handle, work.create_pack()).await.unwrap();
+    handle
+        .publish_push(
+            Some(ingested),
+            make_txn(vec![("refs/heads/main", "", &c1)]),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(handle.local().refs().unwrap().refs.len(), 1);
+
+    store
+        .delete(
+            &format!("{}{}", id.store_prefix(), walgit_proto::keys::MANIFEST),
+            None,
+        )
+        .await
+        .unwrap();
+
+    let in_flight = handle.sync_refs().await.unwrap();
+    let mut create = tokio::spawn({
+        let registry = registry.clone();
+        let id = id.clone();
+        async move { registry.create(&id, ObjectFormat::Sha1).await }
+    });
+
+    // The create retires the handle before it waits; a request started now is refused.
+    loop {
+        match handle.sync_refs().await {
+            Err(WalError::NotFound) => break,
+            Ok(guard) => drop(guard),
+            Err(e) => panic!("unexpected {e}"),
+        }
+        tokio::task::yield_now().await;
+    }
+    assert!(
+        tokio::time::timeout(std::time::Duration::from_millis(100), &mut create)
+            .await
+            .is_err(),
+        "create must wait for the request still holding the old handle"
+    );
+    assert_eq!(handle.local().refs().unwrap().refs.len(), 1);
+
+    drop(in_flight);
+    let recreated = create.await.unwrap().unwrap();
+
+    assert!(matches!(handle.sync().await, Err(WalError::NotFound)));
+    let late = handle
+        .publish_push(
+            None,
+            make_txn(vec![("refs/heads/late", "", &c1)]),
+            HashMap::new(),
+        )
+        .await;
+    assert!(late.is_err());
+    assert!(recreated.local().refs().unwrap().refs.is_empty());
+    let opened = registry.open(&id).await.unwrap();
+    assert!(Arc::ptr_eq(&opened, &recreated));
+    assert!(!Arc::ptr_eq(&opened, &handle));
+}
+
+/// A handle dropped by `delete` is retired too, so a holder that outlives the delete
+/// cannot write into the directory of a repository created afterwards.
+#[tokio::test]
+async fn a_handle_held_across_delete_and_create_cannot_touch_the_new_repository() {
+    let cache = tempfile::tempdir().unwrap();
+    let store = MemoryStore::shared();
+    let registry = Registry::new(
+        store.clone(),
+        Arc::new(make_config(&cache.path().join("a"), 0)),
+    );
+    let id = repo_id("test", "deleted");
+
+    let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap();
+    let work = WorkRepo::new();
+    let c1 = work.commit("init", "data");
+    let ingested = ingest_pack_data(&handle, work.create_pack()).await.unwrap();
+    handle
+        .publish_push(
+            Some(ingested),
+            make_txn(vec![("refs/heads/main", "", &c1)]),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+
+    registry.delete(&id).await.unwrap();
+    let recreated = registry.create(&id, ObjectFormat::Sha1).await.unwrap();
+
+    assert!(matches!(handle.sync().await, Err(WalError::NotFound)));
+    let late = handle
+        .publish_push(
+            None,
+            make_txn(vec![("refs/heads/late", "", &c1)]),
+            HashMap::new(),
+        )
+        .await;
+    assert!(late.is_err());
+    assert!(recreated.local().refs().unwrap().refs.is_empty());
+}
+
+/// A Create whose reply was lost may have landed, so the warm copy of a repository another
+/// host deleted must not keep serving its refs under the new manifest.
+#[tokio::test]
+async fn a_create_with_a_lost_reply_stops_serving_the_cached_copy() {
+    use walgit_store::fault::{FaultPlan, FaultStore};
+    let cache = tempfile::tempdir().unwrap();
+    let truth = MemoryStore::shared();
+    let link = FaultStore::new(truth.clone(), "lost-create", 1);
+    let registry = Registry::new(
+        link.clone(),
+        Arc::new(make_config(&cache.path().join("a"), 0)),
+    );
+    let id = repo_id("test", "lost-create");
+
+    let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap();
+    let work = WorkRepo::new();
+    let c1 = work.commit("init", "data");
+    let ingested = ingest_pack_data(&handle, work.create_pack()).await.unwrap();
+    handle
+        .publish_push(
+            Some(ingested),
+            make_txn(vec![("refs/heads/main", "", &c1)]),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+
+    Registry::new(
+        truth.clone(),
+        Arc::new(make_config(&cache.path().join("b"), 0)),
+    )
+    .delete(&id)
+    .await
+    .unwrap();
+
+    link.set(FaultPlan {
+        p_err_after: 1.0,
+        only_keys: Some(vec!["manifest.pb".into()]),
+        ..Default::default()
+    });
+    assert!(registry.create(&id, ObjectFormat::Sha1).await.is_err());
+    link.heal();
+
+    assert!(matches!(handle.sync().await, Err(WalError::NotFound)));
+    let reopened = registry.open(&id).await.unwrap();
+    assert!(reopened.local().refs().unwrap().refs.is_empty());
+}
+
+/// A reset that died after invalidating the directory leaves refs and packs behind;
+/// `open` must rebuild from empty instead of serving them under the new manifest.
+#[tokio::test]
+async fn open_discards_a_directory_whose_reset_was_interrupted() {
+    let cache = tempfile::tempdir().unwrap();
+    let store = MemoryStore::shared();
+    let registry = Registry::new(
+        store.clone(),
+        Arc::new(make_config(&cache.path().join("a"), 0)),
+    );
+    let id = repo_id("test", "interrupted");
+
+    let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap();
+    let work = WorkRepo::new();
+    let c1 = work.commit("init", "data");
+    let ingested = ingest_pack_data(&handle, work.create_pack()).await.unwrap();
+    handle
+        .publish_push(
+            Some(ingested),
+            make_txn(vec![("refs/heads/main", "", &c1)]),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+    let dir = handle.local().path().to_path_buf();
+    drop(handle);
+
+    store
+        .delete(
+            &format!("{}{}", id.store_prefix(), walgit_proto::keys::MANIFEST),
+            None,
+        )
+        .await
+        .unwrap();
+    let other = Registry::new(
+        store.clone(),
+        Arc::new(make_config(&cache.path().join("b"), 0)),
+    );
+    other.create(&id, ObjectFormat::Sha1).await.unwrap();
+    std::fs::remove_file(dir.join("HEAD")).unwrap();
+
+    let fresh = Registry::new(store, Arc::new(make_config(&cache.path().join("a"), 0)));
+    let opened = fresh.open(&id).await.unwrap();
+    assert!(opened.local().refs().unwrap().refs.is_empty());
+}
+
+/// Concurrent creates of one name — on one registry and across two — have
+/// exactly one winner; `open_or_create` (auto-create on push) treats losing
+/// that race as success and opens the winner's repository.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn concurrent_creates_have_one_winner_and_open_or_create_joins_it() {
+    let cache = tempfile::tempdir().unwrap();
+    let store = MemoryStore::shared();
+    let a = Registry::new(
+        store.clone(),
+        Arc::new(make_config(&cache.path().join("a"), 0)),
+    );
+    let b = Registry::new(
+        store.clone(),
+        Arc::new(make_config(&cache.path().join("b"), 0)),
+    );
+
+    for (n, (x, y)) in [(&a, &a), (&a, &b)].into_iter().enumerate() {
+        let id = repo_id("race", &format!("create{n}"));
+        let (rx, ry) = tokio::join!(
+            x.create(&id, ObjectFormat::Sha1),
+            y.create(&id, ObjectFormat::Sha1)
+        );
+        let won = [rx.is_ok(), ry.is_ok()];
+        assert_eq!(won.iter().filter(|w| **w).count(), 1, "{n}: one winner");
+        for r in [rx, ry] {
+            if let Err(e) = r {
+                assert!(matches!(e, walgit_wal::WalError::AlreadyExists), "{n}: {e}");
+            }
+        }
+
+        let id = repo_id("race", &format!("auto{n}"));
+        let (hx, hy) = tokio::join!(
+            x.open_or_create(&id, ObjectFormat::Sha1),
+            y.open_or_create(&id, ObjectFormat::Sha1)
+        );
+        assert_eq!(hx.unwrap().manifest().revision, 1, "{n}");
+        assert_eq!(hy.unwrap().manifest().revision, 1, "{n}");
+    }
+}
+
 #[tokio::test]
 async fn test_concurrent_different_refs() {
     let cache = tempfile::tempdir().unwrap();
diff --git a/docs/EVENTS.md b/docs/EVENTS.md
index 6ca6b97f..8f52b709 100644
--- a/docs/EVENTS.md
+++ b/docs/EVENTS.md
@@ -97,7 +97,9 @@ Wake-ups (both idempotent; they only ever call `catch_up`):
   itself as the notification. Accepted bodies: a GCS Pub/Sub push envelope (`message.attributes.eventType =
   OBJECT_FINALIZE`, `objectId`), an S3 event notification (`Records[].eventName = ObjectCreated:*`,
   `s3.object.key`; MinIO, rustfs and Ceph emit the same shape), or your own glue's `{"key": "repos/o/r/manifest.pb"}`
-  / `{"repo": "o/r"}`. Everything else is acked and ignored; a webhook failure answers 503 so the notifier
+  / `{"repo": "o/r"}`. Keys are bucket object keys as notifications carry them, `store.prefix` included
+  (`{"key": "<prefix>/repos/o/r/manifest.pb"}`); `{"repo": …}` names the repository whatever the prefix.
+  Everything else is acked and ignored; a webhook failure answers 503 so the notifier
   redelivers. Authenticated like every route (`require_read`): give the notifier a token.
 - The sweep (`events.sweep_interval`, default 5 min): `list` + one conditional manifest GET per repo. Not needed
   for correctness; it is the backstop *and the health check* — a sweep that publishes anything means
diff --git a/docs/ROUNDTRIPS.md b/docs/ROUNDTRIPS.md
index bed87112..513d7cb0 100644
--- a/docs/ROUNDTRIPS.md
+++ b/docs/ROUNDTRIPS.md
@@ -55,6 +55,7 @@ right shape. This document is the thinking tool; apply it to every protocol chan
 | Manifest CAS answered 412 | the same evidence check before dropping the own segment: a 412 is contention, or a client library retrying a write that already landed (S3 SDK standard mode, GCS uploads with a precondition). The segment is dropped only when the fresh manifest lists neither it nor a checkpoint at or past its seq (a fold drops descriptors); folded or otherwise unproven evidence stays unknown and keeps the segment | +1 manifest GET on the 412 path only (+1 log GET when our descriptor is listed); successful publishes unchanged | `publish.rs::slot_evidence`, `process_batch`, `publish_compact_classified`, `publish_settings_impl`, `classification.rs::seal_pack_replacement` |
 | Settings publish (D24) | refs sync (conditional GET) → log slot PUT → manifest CAS; readers pay nothing extra (settings ride inline on the manifest) | 3 rounds; read: 0 | `publish.rs::publish_settings_impl` |
 | Lease acquire | 1 GET → 1 CAS put (or 1 Create when absent) | 2 | `coord.rs::try_acquire` |
+| Repository create (`PUT /{o}/{r}`, `walgit repo create`, auto-create on push) | 1 manifest Create PUT; its 412 *is* "exists" (409), whether this instance holds a handle or not; auto-create's lost race adds the open it would have done anyway | 1 (was 0 for a warm handle, which answered 201 for an existing repository) | `registry.rs::create`, `open_or_create` |
 | Publish, local commit (2026-08-23) | unchanged in round trips: after the manifest CAS the ref txns are applied to the local copy **before** the new manifest version is advertised, both under `sync_mutex` (the refs phase of every sync); the reverse order let a reader cache the old refs under the new version, and without the lock a concurrent sync replayed the same entry (two `update-ref`, a lock collision). A landed CAS is answered `ok` whatever the local apply does — the next sync replays (one conditional GET that then returns 200, no extra write). | 0 extra | `publish.rs::process_batch` |
 | Repository listing (`/api/v1/owners*`, `/services/api/owners*`, maintainer/bridge passes) | 0 within `LIST_TTL` (30 s, per instance); else delimited `repos/` → (delimited `repos/<o>/` ∥ owners) → (HEAD `manifest.pb` ∥ repos): 3 rounds | 1 + owners + repos | `registry.rs::list` |
 | Bundle removal (2026-09-11) | v2 capabilities and narrated fetch: removed optional list GET (1 → 0 extra); maintenance no longer reads/CASes a bundle list; direct import no longer composes a wrapper or reads/CASes a bundle list | no new store requests; checkpoint and push budgets unchanged | `smart.rs`, `maintain.rs`, `import_direct.rs` |
@@ -62,7 +63,8 @@ right shape. This document is the thinking tool; apply it to every protocol chan
 | Orphan log slot (failure path only) | +1 fresh manifest GET, +HEAD per probe, +Create at next seq | — | `publish.rs::claim_log_slot` |
 
 `healthy_request_round_trip_budgets` in `crates/walgit-server/tests/sim.rs` pins the healthy MemoryStore
-counts at push **5**, warm refs **1**, cold refs with one tail segment **2**, and checkpoint **4**. Cold open used to spend an
+counts at push **5**, warm refs **1**, cold refs with one tail segment **2**, checkpoint **4**, and create of an
+existing repository **1** (warm and cold). Cold open used to spend an
 extra unconditional manifest GET (3 requests, 3 sequential rounds); it now applies the manifest it already
 fetched directly (2 requests, 2 rounds). `claim_log_slot`, `cas_landed`, and `put_immutable_create` add probes
 only after Create/CAS failure, so the measured happy-path counts remain unchanged.
diff --git a/justfile b/justfile
index c7398eda..64b4e580 100644
--- a/justfile
+++ b/justfile
@@ -10,7 +10,7 @@ t15 := `if command -v timeout >/dev/null 2>&1; then echo "timeout 900"; elif com
 
 # The fast tier's package selections, shared by the build and the run of each line.
 fast_pkgs := "-p walgit-store -p walgit-git -p walgit-wal"
-server_fast := "-p walgit-server --test web_api --test web_ui --test api_v1 --test static_http --test packfile_uri --test forward --test maintain --test routing_prefix --test lfs_upstream --test drain --test events --test follow --test policy"
+server_fast := "-p walgit-server --test web_api --test web_ui --test api_v1 --test static_http --test packfile_uri --test forward --test maintain --test routing_prefix --test metrics --test lfs_upstream --test drain --test events --test follow --test policy"
 
 # Default: show available targets.
 default: