diff --git a/.github/workflows/git-protocol-v2-partial-clone.yml b/.github/workflows/git-protocol-v2-partial-clone.yml
index 827223172..554748afd 100644
--- a/.github/workflows/git-protocol-v2-partial-clone.yml
+++ b/.github/workflows/git-protocol-v2-partial-clone.yml
@@ -45,6 +45,8 @@ on:
       - "crab/tests/**"
       - "crates/crab-git/**"
       - "crates/crab-auth-server/**"
+      - "crates/crab-auth/**"
+      - "crates/crab-staging/**"
       - "crates/crab-metadata/**"
       - "crates/crab-read/**"
       - "crates/crab-remote-git/**"
@@ -99,6 +101,8 @@ on:
       - "crab/tests/**"
       - "crates/crab-git/**"
       - "crates/crab-auth-server/**"
+      - "crates/crab-auth/**"
+      - "crates/crab-staging/**"
       - "crates/crab-metadata/**"
       - "crates/crab-read/**"
       - "crates/crab-remote-git/**"
@@ -176,6 +180,11 @@ jobs:
           cargo test -p crab --bin crab pull_machine_errors_keep_the_selected_mode_and_schema --locked --features gix-transport
           cargo test -p crab --lib git::push_native::tests --locked --features gix-transport
           cargo test -p crab --lib git::push_staging::tests --locked --features gix-transport
+          cargo test -p crab-staging --locked --test push_snapshot_guard --test publication_history_recovery
+          cargo test -p crab-auth --lib protected_push --locked --features crab-auth-client
+          cargo test -p crab --lib git::index::tests --locked --features gix-transport
+          cargo test -p crab --lib publication_recovery --locked --features gix-transport
+          cargo test -p crab --lib git::capsule_push::tests --locked --features gix-transport
           cargo test -p crab --lib --locked --features gix-transport -- \
             git::push::tests::lookup_staging \
             git::push::tests::under_lock_refresh \
@@ -269,6 +278,11 @@ jobs:
           cargo test -p crab --lib cmd::lfs::install::tests --locked --no-default-features --features simd-accel,tier,watch,nfs,gix-pathmatch,gix-transport
           cargo test -p crab --lib git::push_native::tests --locked --no-default-features --features simd-accel,tier,watch,nfs,gix-pathmatch,gix-transport
           cargo test -p crab --lib git::push_staging::tests --locked --no-default-features --features simd-accel,tier,watch,nfs,gix-pathmatch,gix-transport
+          cargo test -p crab-staging --locked --test push_snapshot_guard --test publication_history_recovery
+          cargo test -p crab-auth --lib protected_push --locked --features crab-auth-client
+          cargo test -p crab --lib git::index::tests --locked --no-default-features --features simd-accel,tier,watch,nfs,gix-pathmatch,gix-transport
+          cargo test -p crab --lib publication_recovery --locked --no-default-features --features simd-accel,tier,watch,nfs,gix-pathmatch,gix-transport
+          cargo test -p crab --lib capsule_push_retires_staging_only_after_confirmed_ref_publication --locked --no-default-features --features simd-accel,tier,watch,nfs,gix-pathmatch,gix-transport
           cargo test -p crab --lib git::remote_helper::tests::list_session_does_not_open_a_damaged_staging_index --locked --no-default-features --features simd-accel,tier,watch,nfs,gix-pathmatch,gix-transport
 
       - name: Run upload-pack session contracts
diff --git a/crab/docs/architecture/indexed-remote-reads.md b/crab/docs/architecture/indexed-remote-reads.md
index a10a26ba6..1968be230 100644
--- a/crab/docs/architecture/indexed-remote-reads.md
+++ b/crab/docs/architecture/indexed-remote-reads.md
@@ -2016,3 +2016,117 @@ therefore do not qualify capsule lifecycle. A separate follow-up must bind exact
 selected owners before publication and release only proven committed owners,
 including cancellation, partial/ref-CAS failure and concurrent readers. Until that
 is implemented and tested, local post-push reclamation is not qualified.
+
+### Capsule snapshot lifecycle follow-up
+
+Capsule push fences retirement before selecting local recipes, pins selected
+published/delegated owners, and retains that snapshot through upload and ref
+publication. Confirmed remote commit precedes the local committed journal and
+exact retirement. Local cleanup errors leave maintenance work and preserve the
+successful remote result. Ref rejection or failure retains unpublished leases.
+
+`StagingPushSnapshot` owns marker/snapshot cleanup on drop, including task abort.
+Sibling readers defer retirement to exclusive maintenance. It shares the existing
+index and physical-payload implementation with the released asynchronous API;
+there is no storage-format or dependency change. Guard tests cover drop, task
+abort, shared native ownership, sibling deferral and local journal failure.
+
+Native publication-intent recovery now runs in capsule push before snapshot
+admission. Recovery uses Git's immediate index lock. If the complete prior index
+state is restored, a bounded-memory local Git object walk checks refs and reflogs
+before native rollback. Recipes still referenced there acquire canonical history
+owners before native leases are released. Cancellation or an incomplete history
+walk fails without authorizing rollback. Ordinary pushes without unresolved
+intents do not perform this history scan; this does not replace the historical
+remote-catalog indexing workstream.
+
+The minimum Git 2.30 dependency contract supports the selected walk flags and
+includes other worktrees with `--all --reflog` by default
+([revision options](https://github.com/git/git/blob/v2.30.0/Documentation/rev-list-options.txt)).
+Its transport allowlist takes precedence over configured transport permissions
+([transport implementation](https://github.com/git/git/blob/v2.30.0/transport.c));
+the empty allowlist prevents an implicit fetch during local recovery. Windows
+startup sets stdout to binary mode, matching the OID parser's LF delimiter
+([MinGW startup](https://github.com/git/git/blob/v2.30.0/compat/mingw.c)).
+
+Both exclusive native-add recovery and shared push recovery use this ownership
+rule. The shared staging helper validates all requested recipes against the
+intent before adding history owners, preserves the original intent across an
+interrupted prefix, and permits an idempotent retry. Prepared-only recovery tests
+cover subset retention, full reconstruction and physical xorb deletion only after
+confirmed snapshot commit.
+
+Both protected finalize clients use the shared auth response validator. A response
+with coordinator metadata must attest `Committed` or `Materialized` before it can
+permit local retirement. `Pending`, `ObjectsUploaded` and `Aborted` are rejected;
+the existing successful response without coordinator metadata remains supported.
+The minimal protected-push suite passes 30 tests.
+
+The existing managed-file smoke checks post-push owners, snapshots, raw payloads
+and prepared ownership bytes for native and Git helper push. The frozen #566 CLI
+baseline completes native add/commit/push and verifies the remote tip, then fails
+the retirement assertion with retained owners/prepared payload. Corrected CLI
+proof and final-source 10-GiB qualification remain required.
+
+The large-file harness adds `--pattern entropy|repeated|zero`; entropy preserves
+its original seed/default. Repeated input uses a deterministic one-MiB block;
+zero input writes real blocks. All patterns retain four alternating native/Git
+versions, beginning/middle/end edits, two-file identity checks, source removal and
+cold historical reconstruction. Syntax checks pass; the new patterns are unrun.
+
+The frozen #570 release build succeeds with `simd-accel,gix-pathmatch` and no
+default features. Its binary hash is recorded externally. Resumed workload
+`unified-ingestion-10g-r3` passes the formerly failing v1 Git-clean push and the
+subsequent v2 native push, with 2,216,109 and 2,133,870 incremental xorb bytes.
+The full workload is still running; cold historical reconstruction is unproven.
+This artifact precedes the lifecycle and finalize-contract changes above.
+
+
+The 10-GiB harness also selects `--first-ingestion native|git`, covering both
+native→clean→native and clean→native→clean. Receipts include the selected order
+and each add's ingestion path. After each add, the harness rejects local
+segment/prepared storage (including temporary stream bodies) beyond one initial
+representation, cumulative edited
+bytes and metadata allowance (10% of source size, at least 32 MiB). This detects
+the observed extra full 10-GiB representation; it is not a latency target or a
+claim that all allowed overhead is necessary. Existing runs retain their
+original harness identity and cannot prove these newly added checks.
+
+The local history dependency probe additionally reproduces exit zero with a
+missing reflog commit and an omitted expected pointer. Git reports that partial
+walk only through diagnostics; its
+[reflog traversal implementation](https://github.com/git/git/blob/v2.30.0/revision.c)
+confirms that contract. Recovery therefore requires a diagnostic-free successful
+walk before accepting absence. A warning retains the unresolved intent and its
+payloads; there is no log-message matching or alternate deletion path. Both a
+missing reflog tip and a missing referenced tree are regression cases. Earlier
+consumer builds do not qualify this tightened scanner until rerun.
+
+An extracted-source probe compiles the exact helper and canonical subprocess
+owner against the frozen release's actual error/dependency types. The old helper
+accepts the incomplete walk (negative runtime exits 1); the tightened helper
+preserves the healthy reflog result and rejects the missing tip (positive runtime
+exits 0). Source hashes are recorded with both probes. This is function-level
+runtime proof; the full updated consumer still needs its own Cargo/CLI run.
+
+
+Reduced crossover qualification passes with the frozen #570 release: Git-first
+repeated data and native-first zero data, each with two 128-MiB files, four
+versions, source removal, eight cold file digests and remote fsck. Each run records
+71 commands and 53 passing checks. An older debug binary fails the new storage
+assertion after its first native-to-clean switch: 67,156,162 prepared bytes plus
+68,224,532 segment bytes exceed the 102,760,448-byte bound for its 64-MiB input.
+These runs qualify the order/storage oracle; they do not qualify the final
+lifecycle source or replace the full 10-GiB matrix. The later inclusion of
+stream-temporary bytes in the accounting remains part of that final matrix.
+
+The dedicated #572 provider run passes its provider contract and the initial
+native cutover push's retirement check, then stops at the old missing-manifest
+fixture. It deleted a retired key while `v2/root` remained authoritative. Missing
+and corrupt fixtures now initialize a healthy repository, attest the live root,
+and verify their deletion/overwrite took effect before requiring list/clone to
+fail. Corruption still requires the integrity error; expectations were not
+relaxed. Both updated fixtures pass against the unchanged #570 root consumer
+(23 commands, 23 checks), and the 11 source-isolation tests pass. The explicit
+v1 upgrade/reset fixtures remain separate. Full final-source canary proof is
+still pending.
diff --git a/crab/scripts/e2e/run_add_commit_push_rustfs_smoke.py b/crab/scripts/e2e/run_add_commit_push_rustfs_smoke.py
index 82c378e42..17f2f63d9 100755
--- a/crab/scripts/e2e/run_add_commit_push_rustfs_smoke.py
+++ b/crab/scripts/e2e/run_add_commit_push_rustfs_smoke.py
@@ -8,7 +8,7 @@
 
 The large-file cases prove pointer staging, object-store publication, fresh
 clone, hydration, and byte identity. A separate ordinary-Git matrix proves
-missing/corrupt manifest handling, update/fetch, deletion, force and
+missing/corrupt metadata-root handling, update/fetch, deletion, force and
 non-fast-forward behavior, atomic rejection, tag following, shallow history,
 immutable-pack failures, and concurrent push serialization.
 
@@ -850,19 +850,25 @@ def clone_git(
             check=check,
         )
 
-    def run_missing_manifest_case(self) -> None:
-        case_name = "missing-manifest"
+    def run_missing_root_case(self) -> None:
+        case_name = "missing-root"
         _repo, remote_url, repo_prefix = self.prepare_git_repo(case_name)
+        root_key = f"{repo_prefix}/v2/root"
+        self.ls_remote(remote_url, name=f"{case_name} healthy list")
+        self.head_key(root_key)
         self.run_aws(
-            "delete canonical manifest fixture",
+            "delete canonical root fixture",
             [
                 "delete-object",
                 "--bucket",
                 self.args.bucket,
                 "--key",
-                f"{repo_prefix}/manifest",
+                root_key,
             ],
         )
+        status, _, _ = self.signed_s3_request("HEAD", root_key)
+        self.check(f"{case_name}-fixture-removed", status == 404,
+                   {"key": root_key, "status": status})
         listing = self.run_git(
             self.run_root,
             ["ls-remote", remote_url],
@@ -1174,13 +1180,15 @@ def run_shallow_case(self) -> None:
             {"commit_count": full_depth},
         )
 
-    def run_corrupt_manifest_case(self) -> None:
-        case_name = "corrupt-manifest"
-        remote_url, repo_prefix = self.remote_for_case(case_name)
-        manifest_key = f"{repo_prefix}/manifest"
+    def run_corrupt_root_case(self) -> None:
+        case_name = "corrupt-root"
+        _repo, remote_url, repo_prefix = self.prepare_git_repo(case_name)
+        root_key = f"{repo_prefix}/v2/root"
+        self.ls_remote(remote_url, name=f"{case_name} healthy list")
+        self.head_key(root_key)
         status, _, response_body = self.signed_s3_request(
             "PUT",
-            manifest_key,
+            root_key,
             body=b'{"version":',
         )
         self.check(
@@ -1193,6 +1201,11 @@ def run_corrupt_manifest_case(self) -> None:
             },
         )
 
+        status, _, body = self.signed_s3_request("GET", root_key)
+        self.check(f"{case_name}-fixture-overwritten",
+                   status == 200 and body == b'{"version":',
+                   {"key": root_key, "status": status})
+
         ls_remote = self.run_git(
             self.run_root,
             ["ls-remote", remote_url],
@@ -1607,6 +1620,20 @@ def run_case(self, case_name: str, use_crab_add: bool) -> None:
             )
 
         self.head_repository_state(repo_prefix)
+        ownership = self.run_crab(
+            repo, ["staging", "stats", "--json"],
+            name=f"{case_name} staging ownership after push",
+        )
+        lifecycle = json.loads(self.read_stdout(ownership))["data"]["lifecycle"]
+        self.check(
+            f"{case_name}-retires-published-staging-owners",
+            all(lifecycle[field] == 0 for field in (
+                "path_heads", "path_leases", "open_push_snapshots",
+                "committed_push_snapshots", "payloads",
+                "current_head_prepared_bytes", "snapshot_pinned_prepared_bytes",
+            )),
+            lifecycle,
+        )
         after_xorbs = self.list_keys(".crab/xorbs/")
         after_shards = self.list_keys(".crab/shards/")
         new_xorbs = len(after_xorbs - before_xorbs)
@@ -2193,8 +2220,8 @@ def run(self) -> None:
             self.write_report()
             return
         self.run_v1_hard_cutover_reset_case()
-        self.run_missing_manifest_case()
-        self.run_corrupt_manifest_case()
+        self.run_missing_root_case()
+        self.run_corrupt_root_case()
         self.run_ref_update_delete_force_and_tag_case()
         self.run_atomic_rejection_case()
         self.run_shallow_case()
diff --git a/crab/scripts/e2e/run_large_file_edits_rustfs.py b/crab/scripts/e2e/run_large_file_edits_rustfs.py
index 71323b5bd..4b24ea36d 100644
--- a/crab/scripts/e2e/run_large_file_edits_rustfs.py
+++ b/crab/scripts/e2e/run_large_file_edits_rustfs.py
@@ -1,7 +1,9 @@
 #!/usr/bin/env python3
 """Qualify repeated 10-GiB edits through both Git clean and native Crab push.
 
-Two copy-on-write files start with identical, non-sparse deterministic entropy.
+Two copy-on-write files start with identical, non-sparse deterministic content.
+Choose entropy, a repeated one-MiB block, or zero-filled bytes.
+Choose native-first or Git-first alternating ingestion.
 Edits diverge at the beginning, middle and end. Each committed version must
 hydrate byte-identically from a fresh cache after the source repository is removed.
 Only this invocation's disposable data is released; logs and receipts remain.
@@ -77,9 +79,15 @@ def qualify(runner: AddCommitPushSmoke, proxy: RequestCountingProxy) -> None:
     repo, remote, prefix = runner.prepare_repo("source")
     cap_cache(repo)
     paths = [repo / "model-a.bin", repo / "model-b.bin"]
+    repeated_block = None
+    if args.pattern == "zero":
+        repeated_block = bytes(MIB)
+    elif args.pattern == "repeated":
+        repeated_block = hashlib.shake_256(b"large-edits:repeated").digest(MIB)
     with paths[0].open("wb") as output:
         for block in range(args.file_mib):
-            output.write(hashlib.shake_256(f"large-edits:{block}".encode()).digest(MIB))
+            output.write(repeated_block if repeated_block is not None else
+                         hashlib.shake_256(f"large-edits:{block}".encode()).digest(MIB))
     clone_flag = "-c" if sys.platform == "darwin" else "--reflink=always"
     runner.run_cmd("copy-on-write second large file",
                    ["cp", clone_flag, str(paths[0]), str(paths[1])], repo)
@@ -89,13 +97,15 @@ def qualify(runner: AddCommitPushSmoke, proxy: RequestCountingProxy) -> None:
         (code / f"module-{index:05}.rs").write_text(f"pub const VALUE: u64 = {index};\n")
     runner.check("workload", all(path.stat().st_size == size for path in paths),
                  {"file_bytes": size, "files": 2, "initial_unique_bytes": size,
-                  "versions": 4, "code_files": args.code_files})
+                  "versions": 4, "code_files": args.code_files, "pattern": args.pattern,
+                  "first_ingestion": args.first_ingestion})
     history = []
     phases = []
 
     def record() -> None:
         receipt = runner.artifacts / "large-file-edits.json"
-        receipt.write_text(json.dumps({"history": history, "phases": phases,
+        receipt.write_text(json.dumps({"history": history, "phases": phases, "pattern": args.pattern,
+                                      "first_ingestion": args.first_ingestion,
                                       "transport": proxy.snapshot()}, indent=2) + "\n")
         runner.report.artifacts["large_file_edits"] = str(receipt)
         runner.write_report()
@@ -109,7 +119,8 @@ def record() -> None:
                     stream.write(hashlib.shake_256(f"edit:{version}:{index}".encode()).digest(MIB))
             (code / "module-00000.rs").write_text(f"pub const VERSION: u64 = {version};\n")
         before = proxy.snapshot()
-        if version % 2:
+        native_add = (version % 2 == 0) == (args.first_ingestion == "native")
+        if not native_add:
             add = runner.run_git(repo, ["add", "--", *[p.name for p in paths]],
                                  name=f"v{version} Git clean-filter add")
         else:
@@ -117,9 +128,17 @@ def record() -> None:
         staging = repo / ".crab" / "staging"
         local_bytes = {
             name: sum(path.stat().st_size for path in (staging / name).rglob("*") if path.is_file())
-            for name in ("segments", "push-plans")
+            for name in ("segments", "push-plans", "stream-prepared-xorbs")
         }
+        # One initial representation plus all edited bytes is sufficient. Allow
+        # encoded metadata overhead, while rejecting the observed second full copy.
+        max_local_bytes = size + 2 * version * MIB + max(size // 10, 32 * MIB)
+        runner.check(f"v{version}-no-second-full-local-payload",
+                     sum(local_bytes.values()) <= max_local_bytes,
+                     {"staging_directory_bytes": local_bytes,
+                      "max_bytes": max_local_bytes})
         phases.append({"version": version, "phase": "add", "duration_ms": add.duration_ms,
+                       "ingestion": "native" if native_add else "git",
                        "staging_directory_bytes": local_bytes,
                        "transport": RequestCountingProxy.delta(before, proxy.snapshot())})
         for path in paths:
@@ -194,6 +213,8 @@ def main() -> None:
     parser.add_argument("--endpoint-url", required=True)
     parser.add_argument("--file-mib", type=int, default=10240)
     parser.add_argument("--code-files", type=int, default=5000)
+    parser.add_argument("--pattern", choices=("entropy", "repeated", "zero"), default="entropy")
+    parser.add_argument("--first-ingestion", choices=("native", "git"), default="native")
     args = parser.parse_args()
     if args.file_mib < 16 or args.code_files < 1 or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", args.run_id):
         parser.error("require >=16 MiB/file, positive code files and a safe run id")
diff --git a/crab/src/cmd/add.rs b/crab/src/cmd/add.rs
index 006dc45b8..5dc8759fc 100644
--- a/crab/src/cmd/add.rs
+++ b/crab/src/cmd/add.rs
@@ -1050,7 +1050,7 @@ async fn execute_add(
     // pointers published by the process that just released the staging lock.
     let staging_root = worktree_ctx.shared_staging_dir();
     let lock_wait_start = Instant::now();
-    let staging = Arc::new(open_add_staging(&staging_root, &repo_root).await?);
+    let staging = Arc::new(open_add_staging(&staging_root, &repo_root, cancel).await?);
     let lock_wait_duration_ms = lock_wait_start.elapsed().as_millis() as u64;
     let (candidates, clean_skipped) =
         filter_clean_indexed_candidates(&repo_root, candidates, jobs, cancel).await?;
@@ -1656,12 +1656,16 @@ async fn execute_add(
     Ok(summary)
 }
 
-async fn open_add_staging(staging_root: &Path, repo_root: &Path) -> Result<StagingArea> {
+async fn open_add_staging(
+    staging_root: &Path,
+    repo_root: &Path,
+    cancel: &CancellationToken,
+) -> Result<StagingArea> {
     let staging =
         StagingArea::open_blocking(staging_root.to_path_buf(), ADD_STAGING_LOCK_WAIT_BUDGET)
             .await
             .map_err(CrabError::from)?;
-    reconcile_publication_intents_writer(&staging, repo_root)?;
+    reconcile_publication_intents_writer(&staging, repo_root, cancel)?;
     Ok(staging)
 }
 
@@ -5891,7 +5895,8 @@ mod tests {
             let repo_root = repo_root.clone();
             let path = path.clone();
             async move {
-                let staging = open_add_staging(&root, &repo_root).await?;
+                let staging =
+                    open_add_staging(&root, &repo_root, &CancellationToken::new()).await?;
                 let filtered = filter_clean_indexed_candidates(
                     &repo_root,
                     vec![(path, data.len() as u64)],
@@ -6225,7 +6230,9 @@ mod tests {
         assert!(intent.is_some());
         staging.close().await.unwrap();
 
-        let recovered = open_add_staging(&staging_root, &repo).await.unwrap();
+        let recovered = open_add_staging(&staging_root, &repo, &CancellationToken::new())
+            .await
+            .unwrap();
 
         assert!(
             recovered
@@ -6271,7 +6278,7 @@ mod tests {
         staging.close().await.unwrap();
 
         let recovered = StagingAreaReadOnly::open(staging_root).await.unwrap();
-        reconcile_publication_intents(&recovered, &repo)
+        reconcile_publication_intents(&recovered, &repo, &CancellationToken::new())
             .await
             .unwrap();
 
@@ -6323,14 +6330,19 @@ mod tests {
             let (error, intents, bytes) = if read_only {
                 staging.close().await.unwrap();
                 let staging = StagingAreaReadOnly::open(root).await.unwrap();
-                let result = reconcile_publication_intents(&staging, &repo).await;
+                let result =
+                    reconcile_publication_intents(&staging, &repo, &CancellationToken::new()).await;
                 (
                     result.unwrap_err(),
                     staging.unresolved_publication_intents().unwrap().len(),
                     staging.get_chunk(&compute_data_hash(data)).await.unwrap(),
                 )
             } else {
-                let result = reconcile_publication_intents_writer(&staging, &repo);
+                let result = reconcile_publication_intents_writer(
+                    &staging,
+                    &repo,
+                    &CancellationToken::new(),
+                );
                 let state = (
                     result.unwrap_err(),
                     staging.unresolved_publication_intents().unwrap().len(),
@@ -6428,7 +6440,7 @@ mod tests {
         })
         .await
         .unwrap();
-        let blocked = reconcile_publication_intents(&staging, repo)
+        let blocked = reconcile_publication_intents(&staging, repo, &CancellationToken::new())
             .await
             .unwrap_err();
         let retained = staging.unresolved_publication_intents().unwrap().len() == 1
@@ -6450,7 +6462,9 @@ mod tests {
         .await
         .unwrap();
         assert!(status.success(), "Git index installation failed");
-        reconcile_publication_intents(&staging, repo).await.unwrap();
+        reconcile_publication_intents(&staging, repo, &CancellationToken::new())
+            .await
+            .unwrap();
         assert!(
             matches!(blocked, CrabError::Io(_))
                 && retained
@@ -6510,7 +6524,7 @@ mod tests {
         staging.close().await.unwrap();
 
         let recovered = StagingAreaReadOnly::open(staging_root).await.unwrap();
-        let error = reconcile_publication_intents(&recovered, &repo)
+        let error = reconcile_publication_intents(&recovered, &repo, &CancellationToken::new())
             .await
             .expect_err("mixed publication state must fail closed");
 
diff --git a/crab/src/git/capsule_push.rs b/crab/src/git/capsule_push.rs
index 23ae886f9..ae657e236 100644
--- a/crab/src/git/capsule_push.rs
+++ b/crab/src/git/capsule_push.rs
@@ -6,6 +6,7 @@ use std::sync::Arc;
 
 use bytes::Bytes;
 use crab_staging::StagingAreaReadOnly;
+use crab_xet::hash::MerkleHash;
 use gix_object::{Exists, Find, FindHeader};
 use tokio_util::sync::CancellationToken;
 
@@ -322,6 +323,25 @@ async fn run_inner(
         config.force_full_graph,
     )
     .await?;
+    if !prepared.pointers.is_empty()
+        && let Some(staging) = staging
+        && !staging.unresolved_publication_intents()?.is_empty()
+    {
+        // A native add can install Git's pointer before its local publication
+        // acknowledgement. Recover under the canonical index lock before admission.
+        let worktree = gix::open(&git_dir)
+            .map_err(|error| CrabError::Io(std::io::Error::other(error)))?
+            .workdir()
+            .map(Path::to_path_buf)
+            .ok_or_else(|| {
+                CrabError::StagingCorrupt(
+                    "native publication recovery requires a Git working tree".to_owned(),
+                )
+            })?;
+        super::index::reconcile_publication_intents(staging, &worktree, cancel).await?;
+    }
+
+    let staging_snapshot = capture_staging_snapshot(staging, &prepared.pointers)?;
     let visibility_delta = prepare_visibility_delta(
         &common_git_dir,
         &edits,
@@ -519,6 +539,11 @@ async fn run_inner(
         }
     };
     if let Some(active_active_commit) = committed {
+        if let Some(snapshot) = staging_snapshot
+            && let Err(error) = snapshot.commit()
+        {
+            tracing::warn!(%error, "capsule committed; local staging cleanup needs maintenance");
+        }
         for ref_name in ref_names {
             outcomes.insert(ref_name, RefPushOutcome::Ok);
         }
@@ -544,6 +569,41 @@ async fn run_inner(
     Ok((PushResult::new(outcomes), None))
 }
 
+fn capture_staging_snapshot(
+    staging: Option<&Arc<StagingAreaReadOnly>>,
+    pointers: &[crab_types::pointer::Pointer],
+) -> Result<Option<crab_staging::StagingPushSnapshot>> {
+    let Some(staging) = staging.filter(|_| !pointers.is_empty()) else {
+        return Ok(None);
+    };
+    // Fence sibling retirement before selecting local roots; snapshot pins
+    // then protect exactly those leases through upload and the ref transition.
+    let snapshot = crab_staging::StagingPushSnapshot::begin(
+        Arc::clone(staging),
+        uuid::Uuid::now_v7().to_string(),
+    )?;
+    let hashes = pointers
+        .iter()
+        .map(|pointer| MerkleHash::from(pointer.file_hash))
+        .collect::<Vec<_>>();
+    let selected = super::push_staging::recipes_for_selected_pointers(staging, &hashes)?;
+    for pointer in pointers {
+        if let Some(Some(recipe)) = selected.get(&MerkleHash::from(pointer.file_hash))
+            && recipe.file_size() != pointer.size
+        {
+            return Err(CrabError::StagingCorrupt(
+                "selected push recipe disagrees with pointer size".to_owned(),
+            ));
+        }
+    }
+    let recipes = selected.into_values().flatten().collect::<Vec<_>>();
+    if recipes.is_empty() {
+        return Ok(None);
+    }
+    snapshot.capture(&recipes)?;
+    Ok(Some(snapshot))
+}
+
 enum CapsulePublishAttempt {
     Committed(Option<crab_coordination::write_coordinator::CommitOutcome>),
     RefChanged,
@@ -2006,6 +2066,153 @@ mod tests {
         );
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn capsule_push_retires_staging_only_after_confirmed_ref_publication() {
+        use bstr::ByteSlice;
+        use crab_staging::stream::{StreamStageProgress, stage_reader_streaming};
+        let _git_env = crate::test::git_repo::CleanGitEnvGuard::new();
+        for (stale, native_intent, restored_index) in [
+            (false, false, false),
+            (true, false, false),
+            (false, true, false),
+            (true, true, false),
+            (false, true, true),
+        ] {
+            let source = tempfile::tempdir().unwrap();
+            git(source.path(), &["init", "-b", "main"]);
+            git(source.path(), &["config", "user.name", "Crab Test"]);
+            git(
+                source.path(),
+                &["config", "user.email", "crab@example.invalid"],
+            );
+            let first = commit(source.path(), "initial");
+            let staging_root = source.path().join(".crab/staging");
+            let staging = crab_staging::StagingArea::open(staging_root.clone())
+                .await
+                .unwrap();
+            let staged = stage_reader_streaming(
+                &mut std::io::Cursor::new(b"capsule payload"),
+                Path::new("model.bin"),
+                &staging,
+                StreamStageProgress::default(),
+                &CancellationToken::new(),
+            )
+            .await
+            .unwrap();
+            if !native_intent {
+                staging
+                    .delegate_publication(&[(&staged.batch_id, Path::new("model.bin"))])
+                    .unwrap();
+            }
+            let pointer = crab_types::pointer::Pointer {
+                file_hash: staged.file_hash,
+                size: staged.size,
+                shard_hint: None,
+            };
+            std::fs::write(source.path().join("model.bin"), pointer.serialize()).unwrap();
+            if native_intent {
+                let index = gix_index::File::at_or_default(
+                    source.path().join(".git/index"),
+                    gix_hash::Kind::Sha1,
+                    false,
+                    gix_index::decode::Options::default(),
+                )
+                .unwrap();
+                let previous_index_state =
+                    super::super::index::git_index_path_state(&index, b"model.bin".as_bstr());
+                staging
+                    .create_publication_intent(&[crab_staging::PublicationIntentEntry {
+                        batch_id: staged.batch_id.clone(),
+                        path: "model.bin".into(),
+                        expected_pointer_oid: git(
+                            source.path(),
+                            &["hash-object", "-w", "model.bin"],
+                        ),
+                        previous_index_state,
+                    }])
+                    .unwrap();
+            }
+            git(source.path(), &["add", "model.bin"]);
+            git(source.path(), &["commit", "-m", "managed payload"]);
+            let tip = git(source.path(), &["rev-parse", "HEAD"]);
+            if restored_index {
+                git(source.path(), &["read-tree", "HEAD^"]);
+            }
+            staging.close().await.unwrap();
+            let staging = Arc::new(StagingAreaReadOnly::open(staging_root).await.unwrap());
+            let storage = crab_storage::Store::new(Arc::new(InMemory::new()));
+            let store = crate::storage::store::Store::from_storage(storage);
+            let router = crate::storage::StoreLayout::new(store.clone(), "repos/test".to_owned());
+            let layout =
+                crab_storage::StoreLayout::new(store.as_storage().clone(), "repos/test".to_owned());
+            let root = crab_write::capsule_protocol::initialize(
+                &layout,
+                &"1".repeat(64),
+                "refs/heads/main",
+            )
+            .await
+            .unwrap();
+            let limits = crab_read::capsule_protocol::CapsuleReadLimits {
+                max_capsule_bytes: 16 * 1024 * 1024,
+                max_frontier_bytes: 16 * 1024 * 1024,
+            };
+            let initial = crab_read::capsule_protocol::open_view_from_root(&layout, root, limits)
+                .await
+                .unwrap();
+            if stale {
+                publish_ref_edits_with_visibility(
+                    &layout,
+                    &source.path().join(".git"),
+                    vec![crab_metadata::capsule_protocol::CapsuleRefEdit::new(
+                        "refs/heads/main",
+                        None,
+                        Some(first.clone()),
+                        None,
+                    )],
+                )
+                .await;
+            }
+            let config = PushConfig {
+                git_dir: Some(source.path().join(".git")),
+                ..Default::default()
+            };
+            let (result, _) = run(
+                &config,
+                &[PushSpec {
+                    force: false,
+                    src: "refs/heads/main".to_owned(),
+                    dst: "refs/heads/main".to_owned(),
+                }],
+                &store,
+                &router,
+                Some(initial.into()),
+                &[],
+                Some(&staging),
+                None,
+                None,
+                &CancellationToken::new(),
+            )
+            .await
+            .unwrap();
+            let health = staging.lifecycle_health().unwrap();
+            let view = crab_read::capsule_protocol::open_view(&layout, limits)
+                .await
+                .unwrap();
+            assert!(
+                result.all_ok() != stale
+                    && view.refs().get("refs/heads/main")
+                        == Some(if stale { &first } else { &tip })
+                    && health.path_leases == u64::from(stale)
+                    && health.payloads == u64::from(stale)
+                    && health.open_push_snapshots == 0
+                    && health.committed_push_snapshots == 0
+                    && health.unresolved_publications == 0
+                    && staging.list_inflight().unwrap().is_empty(),
+                "stale={stale}, native_intent={native_intent}, restored_index={restored_index}, health={health:?}"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn single_ref_successor_extends_committed_multi_ref_history() {
         let source = tempfile::tempdir().expect("source repository");
diff --git a/crab/src/git/index.rs b/crab/src/git/index.rs
index bba69e0c3..4af4fed25 100644
--- a/crab/src/git/index.rs
+++ b/crab/src/git/index.rs
@@ -1,5 +1,8 @@
+use crab_xet::hash::MerkleHash;
+use std::collections::HashSet;
 use std::path::Path;
 use std::process::Command;
+use tokio_util::sync::CancellationToken;
 
 use crate::core::error::{CrabError, Result};
 use crab_staging::{PublicationIntentId, StagingArea, StagingAreaReadOnly};
@@ -52,20 +55,31 @@ pub(crate) fn git_index_path_state(index: &gix_index::File, path: &bstr::BStr) -
 pub(crate) async fn reconcile_publication_intents(
     staging: &StagingAreaReadOnly,
     repo_root: &Path,
+    cancel: &CancellationToken,
 ) -> Result<()> {
+    crate::core::error::check_cancelled(cancel)?;
     if staging.unresolved_publication_intents()?.is_empty() {
         return Ok(());
     }
     let (_index_lock, index) = lock_publication_index(repo_root)?;
     // Another shared recovery may have completed before lock acquisition.
     // Read intents again under the lock rather than acting on a stale list.
-    let actions = publication_recovery_actions(staging.unresolved_publication_intents()?, &index)?;
+    let actions = publication_recovery_actions(
+        staging.unresolved_publication_intents()?,
+        &index,
+        repo_root,
+        cancel,
+    )?;
     for action in actions {
+        crate::core::error::check_cancelled(cancel)?;
         match action {
             PublicationRecoveryAction::Publish(intent_id) => {
                 staging.publish_publication_intent(&intent_id)?;
             }
-            PublicationRecoveryAction::Rollback(intent_id) => {
+            PublicationRecoveryAction::Rollback(intent_id, history) => {
+                if !history.is_empty() {
+                    staging.preserve_publication_history(&intent_id, &history, cancel)?;
+                }
                 staging.rollback_publication_intent(&intent_id).await?;
             }
         }
@@ -77,20 +91,31 @@ pub(crate) async fn reconcile_publication_intents(
 pub(crate) fn reconcile_publication_intents_writer(
     staging: &StagingArea,
     repo_root: &Path,
+    cancel: &CancellationToken,
 ) -> Result<()> {
+    crate::core::error::check_cancelled(cancel)?;
     if staging.unresolved_publication_intents()?.is_empty() {
         return Ok(());
     }
     let (_index_lock, index) = lock_publication_index(repo_root)?;
     // Another shared recovery may have completed before lock acquisition.
     // Read intents again under the lock rather than acting on a stale list.
-    let actions = publication_recovery_actions(staging.unresolved_publication_intents()?, &index)?;
+    let actions = publication_recovery_actions(
+        staging.unresolved_publication_intents()?,
+        &index,
+        repo_root,
+        cancel,
+    )?;
     for action in actions {
+        crate::core::error::check_cancelled(cancel)?;
         match action {
             PublicationRecoveryAction::Publish(intent_id) => {
                 staging.publish_publication_intent(&intent_id)?;
             }
-            PublicationRecoveryAction::Rollback(intent_id) => {
+            PublicationRecoveryAction::Rollback(intent_id, history) => {
+                if !history.is_empty() {
+                    staging.preserve_publication_history(&intent_id, &history, cancel)?;
+                }
                 staging.rollback_publication_intent(&intent_id)?;
             }
         }
@@ -191,7 +216,7 @@ fn delegated_pointer_matches(
 
 enum PublicationRecoveryAction {
     Publish(PublicationIntentId),
-    Rollback(PublicationIntentId),
+    Rollback(PublicationIntentId, Vec<MerkleHash>),
 }
 
 fn lock_publication_index(repo_root: &Path) -> Result<(gix_lock::Marker, gix_index::File)> {
@@ -226,11 +251,14 @@ fn lock_publication_index(repo_root: &Path) -> Result<(gix_lock::Marker, gix_ind
 fn publication_recovery_actions(
     intents: Vec<crab_staging::UnresolvedPublicationIntent>,
     index: &gix_index::File,
+    repo_root: &Path,
+    cancel: &CancellationToken,
 ) -> Result<Vec<PublicationRecoveryAction>> {
     use bstr::ByteSlice;
     use gix_index::entry;
 
     let mut actions = Vec::with_capacity(intents.len());
+    let mut rollbacks = Vec::new();
     for intent in intents {
         let new_matches = intent
             .entries
@@ -255,7 +283,7 @@ fn publication_recovery_actions(
         if new_matches == intent.entries.len() {
             actions.push(PublicationRecoveryAction::Publish(intent.intent_id));
         } else if old_matches == intent.entries.len() {
-            actions.push(PublicationRecoveryAction::Rollback(intent.intent_id));
+            rollbacks.push(intent);
         } else {
             return Err(CrabError::StagingCorrupt(format!(
                 "add publication {} has an ambiguous Git index state ({new_matches}/{} new paths, {old_matches}/{} prior paths); restore one complete index state, then retry",
@@ -265,14 +293,153 @@ fn publication_recovery_actions(
             )));
         }
     }
+    let wanted = rollbacks
+        .iter()
+        .flat_map(|intent| intent.entries.iter())
+        .map(|entry| entry.expected_pointer_oid.clone())
+        .collect::<HashSet<_>>();
+    let committed = committed_pointer_oids(repo_root, &wanted, cancel)?;
+    for intent in rollbacks {
+        let history = intent
+            .entries
+            .iter()
+            .filter(|entry| committed.contains(&entry.expected_pointer_oid))
+            .map(|entry| entry.recipe_hash)
+            .collect();
+        actions.push(PublicationRecoveryAction::Rollback(
+            intent.intent_id,
+            history,
+        ));
+    }
     Ok(actions)
 }
 
+fn committed_pointer_oids(
+    repo_root: &Path,
+    wanted: &HashSet<String>,
+    cancel: &CancellationToken,
+) -> Result<HashSet<String>> {
+    use std::io::{BufRead, Read};
+    if wanted.is_empty() {
+        return Ok(HashSet::new());
+    }
+    // Only crash recovery needs this local history walk. Stream OIDs from refs
+    // and reflogs: an old index alone cannot authorize deleting committed bytes.
+    let mut command = Command::new("git");
+    command
+        .current_dir(repo_root)
+        .env("GIT_NO_LAZY_FETCH", "1")
+        .env("GIT_ALLOW_PROTOCOL", "")
+        .args([
+            "--no-replace-objects",
+            "rev-list",
+            "--all",
+            "--reflog",
+            "--objects",
+            "--no-object-names",
+            "--use-bitmap-index",
+        ]);
+    for name in super::process::GIT_ENV_REMOVALS {
+        command.env_remove(name);
+    }
+    let output = super::process::run(
+        command,
+        cancel,
+        None::<fn(std::process::ChildStdin) -> Result<()>>,
+        |stdout| {
+            let mut found = HashSet::new();
+            let mut reader = std::io::BufReader::new(stdout);
+            let mut line = String::with_capacity(41);
+            loop {
+                crate::core::error::check_cancelled(cancel)?;
+                line.clear();
+                let count = (&mut reader).take(42).read_line(&mut line)?;
+                if count == 0 {
+                    break;
+                }
+                let oid = line
+                    .strip_suffix('\n')
+                    .filter(|oid| oid.len() == 40 && oid.bytes().all(|b| b.is_ascii_hexdigit()))
+                    .ok_or_else(|| {
+                        CrabError::Internal(
+                            "Git recovery history scan returned an invalid object ID".to_owned(),
+                        )
+                    })?;
+                if wanted.contains(oid) {
+                    found.insert(oid.to_owned());
+                }
+            }
+            Ok(found)
+        },
+    )?;
+    // Git can exit zero while skipping unavailable reflog tips. A diagnostic
+    // means this walk cannot attest absence and authorize payload deletion.
+    if !output.status.success() || !output.stderr.is_empty() {
+        return Err(CrabError::Internal(format!(
+            "Git recovery history scan did not complete cleanly ({}): {}",
+            output.status,
+            String::from_utf8_lossy(&output.stderr).trim()
+        )));
+    }
+    Ok(output.stdout)
+}
+
 #[cfg(test)]
 #[allow(clippy::unwrap_used, reason = "test assertions")]
 mod tests {
     use super::*;
 
+    #[test]
+    fn recovery_history_includes_reflogs_and_rejects_incomplete_walks() {
+        let _env = crate::test::git_repo::CleanGitEnvGuard::new();
+        for revision in ["HEAD^{tree}", "HEAD"] {
+            let temp = tempfile::tempdir().unwrap();
+            let git = |args: &[&str]| {
+                let output = Command::new("git")
+                    .args(args)
+                    .current_dir(temp.path())
+                    .output()
+                    .unwrap();
+                assert!(output.status.success(), "{:?}", output.stderr);
+                String::from_utf8(output.stdout).unwrap().trim().to_owned()
+            };
+            git(&["init", "-q", "-b", "main"]);
+            git(&["config", "user.name", "Crab Test"]);
+            git(&["config", "user.email", "crab@example.invalid"]);
+            std::fs::write(temp.path().join("file"), b"initial").unwrap();
+            git(&["add", "file"]);
+            git(&["commit", "-qm", "initial"]);
+            let initial = git(&["rev-parse", "HEAD"]);
+            std::fs::write(temp.path().join("file"), b"committed pointer").unwrap();
+            git(&["add", "file"]);
+            git(&["commit", "-qm", "pointer"]);
+            let pointer = git(&["rev-parse", "HEAD:file"]);
+            let missing = git(&["rev-parse", revision]);
+            git(&["reset", "--hard", &initial]);
+            let wanted = HashSet::from([pointer]);
+            let cancel = CancellationToken::new();
+            assert_eq!(
+                committed_pointer_oids(temp.path(), &wanted, &cancel).unwrap(),
+                wanted
+            );
+            cancel.cancel();
+            assert!(matches!(
+                committed_pointer_oids(temp.path(), &wanted, &cancel),
+                Err(CrabError::Cancelled)
+            ));
+            std::fs::remove_file(
+                temp.path()
+                    .join(".git/objects")
+                    .join(&missing[..2])
+                    .join(&missing[2..]),
+            )
+            .unwrap();
+            assert!(
+                committed_pointer_oids(temp.path(), &wanted, &CancellationToken::new()).is_err()
+            );
+        }
+    }
+
     #[test]
     fn metadata_staging_propagates_git_index_failure() {
         let dir = tempfile::tempdir().unwrap();
diff --git a/crab/src/git/push.rs b/crab/src/git/push.rs
index f5bb8df89..d9e7b97fe 100644
--- a/crab/src/git/push.rs
+++ b/crab/src/git/push.rs
@@ -9716,6 +9716,7 @@ impl PushPipeline {
             crate::git::index::reconcile_publication_intents(
                 staging,
                 &worktree.current_worktree_root,
+                &self.cancel,
             )
             .await?;
         }
diff --git a/crates/crab-auth/README.md b/crates/crab-auth/README.md
index 56681e7e0..034f1643d 100644
--- a/crates/crab-auth/README.md
+++ b/crates/crab-auth/README.md
@@ -18,6 +18,11 @@ The crate also owns the wire contracts used by protected pushes and managed
 transfer grants. A caller can therefore validate an auth response before any
 credential is handed to an object-store builder.
 
+A successful protected finalize response requires `status: updated`. If it
+includes coordinator metadata, the complete metadata must attest `Committed`
+or `Materialized`; pending, uploaded-only and aborted states are rejected before
+a caller can retire its local staging payloads.
+
 ## Architecture
 
 The main flow is:
diff --git a/crates/crab-auth/src/protected_push.rs b/crates/crab-auth/src/protected_push.rs
index b796f6419..96e855346 100644
--- a/crates/crab-auth/src/protected_push.rs
+++ b/crates/crab-auth/src/protected_push.rs
@@ -162,6 +162,17 @@ pub fn validate_push_finalize_response(response: &PushFinalizeResponse) -> Resul
         return invalid_finalize_response("partial active-active commit metadata");
     }
 
+    // A successful finalize authorizes releasing the client's local payloads.
+    // Pending or aborted coordinator state cannot attest remote publication.
+    if let Some(state) = response.commit_state
+        && !matches!(
+            state,
+            PushTransactionState::Committed | PushTransactionState::Materialized
+        )
+    {
+        return invalid_finalize_response("active-active push has not committed");
+    }
+
     if let Some(operation_id) = response.operation_id.as_deref()
         && operation_id.trim().is_empty()
     {
@@ -560,6 +571,39 @@ mod tests {
         assert!(err.to_string().contains("partial active-active"));
     }
 
+    #[test]
+    fn finalize_success_requires_committed_coordinator_state() {
+        for state in [
+            PushTransactionState::Pending,
+            PushTransactionState::ObjectsUploaded,
+            PushTransactionState::Committed,
+            PushTransactionState::Materialized,
+            PushTransactionState::Aborted,
+        ] {
+            let outcome = CommitOutcome {
+                operation_id: "operation".to_owned(),
+                coordinator_epoch: 1,
+                writer: "writer".to_owned(),
+                region: "region".to_owned(),
+                manifest_generation: 1,
+                commit_sequence: 1,
+                state,
+            };
+            let response = PushFinalizeResponse::updated_with_commit_outcome(
+                vec![valid_ref_update()],
+                Some(&outcome),
+            );
+            assert_eq!(
+                validate_push_finalize_response(&response).is_ok(),
+                matches!(
+                    state,
+                    PushTransactionState::Committed | PushTransactionState::Materialized
+                ),
+                "state={state:?}",
+            );
+        }
+    }
+
     #[test]
     fn validate_push_finalize_response_rejects_unexpected_status() {
         let response = PushFinalizeResponse {
diff --git a/crates/crab-staging/AGENTS.md b/crates/crab-staging/AGENTS.md
index b083d1aa6..3818b2299 100644
--- a/crates/crab-staging/AGENTS.md
+++ b/crates/crab-staging/AGENTS.md
@@ -62,6 +62,11 @@ Trace one path: `crab/src/cmd/add.rs` (`close_staging_before_indexing`) →
   active native owners. Layout v3 upgrades released v1 atomically and rejects the
   unreleased v2 prototype. Sources: `index.rs`, `lib.rs`, and crate README.
 
+- `StagingPushSnapshot` owns reader fencing and exact snapshot cleanup. Product
+  callers establish remote commit; the guard retains deferred committed journals
+  and preserves unpublished leases on cancellation. Read `src/push_snapshot.rs`
+  and the capsule consumer before changing this boundary.
+
 ## Features and platform
 
 No declared Cargo features. Requires writable local filesystem/SQLite and platform lock semantics. Scale tests under `crates/crab-staging/tests` need dedicated disk/time budgets; do not run them as an incidental unit check.
diff --git a/crates/crab-staging/README.md b/crates/crab-staging/README.md
index be12d13db..7b954a0dc 100644
--- a/crates/crab-staging/README.md
+++ b/crates/crab-staging/README.md
@@ -94,6 +94,19 @@ and active native transactions survive. Publishing a newer head never cancels
 another caller's open lease. Failed Git conversion can leave a retained handoff;
 reconciliation and abandoned-handoff reclamation still need qualification.
 
+`StagingPushSnapshot::begin` fences retirement before recipe selection;
+`capture` pins exact selected leases. Call `commit` only after confirmed remote
+publication. It records a committed journal before retiring captured owners.
+Drop clears the reader marker and discards an uncommitted snapshot when no
+sibling reader prevents cleanup. Deferred journals remain for exclusive
+maintenance; native and uncaptured leases remain owned throughout.
+
+Native intent recovery may preserve caller-attested committed recipes under
+canonical history owners before rolling back an old Git index state.
+`preserve_publication_history` accepts only verified recipe roots belonging to
+that unresolved intent. It leaves the original intent intact on cancellation or
+error, so interrupted preservation can be retried before rollback.
+
 ## Recovery contract
 
 Opening writable staging runs recovery before creating or reusing its writer.
diff --git a/crates/crab-staging/src/lib.rs b/crates/crab-staging/src/lib.rs
index 4ffd03668..025bbbc30 100644
--- a/crates/crab-staging/src/lib.rs
+++ b/crates/crab-staging/src/lib.rs
@@ -44,6 +44,7 @@ mod index;
 pub mod metrics;
 pub mod multipart_resume;
 pub mod push_plan;
+mod push_snapshot;
 pub mod recipe;
 mod recovery;
 mod segment;
@@ -69,6 +70,7 @@ mod prop_torn_tail_recovery;
 pub use config::StagingConfig;
 pub use error::{Result, StagingError};
 pub use metrics::StagingMetrics;
+pub use push_snapshot::StagingPushSnapshot;
 pub use stats::{
     CompactionStats, RetireStats, StagingCleanStats, StagingLifecycleHealth, StagingStats,
     StagingVerifyStats,
@@ -256,6 +258,97 @@ fn unresolved_publication_intents(
         .collect()
 }
 
+fn publish_verified_recipe_owner(
+    root: &Path,
+    index: &Mutex<Index>,
+    owner: &Path,
+    recipe: &crate::recipe::FileRecipe,
+    display_path: Option<&str>,
+) -> Result<StagingBatchId> {
+    let batch_id = new_staging_batch_id();
+    let file_hash: [u8; 32] = recipe.file_hash().into();
+    let result = (|| {
+        let guard = lock_index(index)?;
+        guard.insert_batch(batch_id.as_str())?;
+        guard.insert_file(&file_hash, recipe.file_size())?;
+        guard.insert_recipe_lease(
+            batch_id.as_str(),
+            &staging_path_bytes(owner),
+            recipe,
+            RecipeVerification::CallerVerified,
+        )?;
+        if let Some(display_path) = display_path {
+            guard.insert_file_path(&file_hash, display_path)?;
+        }
+        guard.mark_batch_published(batch_id.as_str())
+    })();
+    if let Err(error) = result.and_then(|unleased| remove_indexed_files(root, index, &unleased)) {
+        let unowned = lock_index(index).and_then(|guard| {
+            guard.rollback_batch(batch_id.as_str())?;
+            guard.has_file_owner(&file_hash).map(|owned| !owned)
+        });
+        if matches!(unowned, Ok(true)) {
+            let _ = remove_indexed_files(root, index, &[file_hash]);
+        }
+        return Err(error);
+    }
+    Ok(batch_id)
+}
+
+fn preserve_publication_history(
+    root: &Path,
+    index: &Mutex<Index>,
+    intent_id: &PublicationIntentId,
+    recipe_hashes: &[MerkleHash],
+    cancel: &tokio_util::sync::CancellationToken,
+) -> Result<()> {
+    if cancel.is_cancelled() {
+        return Err(StagingError::Cancelled);
+    }
+    let intent = unresolved_publication_intents(index)?
+        .into_iter()
+        .find(|intent| &intent.intent_id == intent_id)
+        .ok_or_else(|| StagingError::NotFound {
+            path: format!("publication intent {}", intent_id.as_str()),
+        })?;
+    let members = intent
+        .entries
+        .iter()
+        .map(|entry| entry.recipe_hash)
+        .collect::<std::collections::HashSet<_>>();
+    let recipes = {
+        let guard = lock_index(index)?;
+        recipe_hashes
+            .iter()
+            .copied()
+            .collect::<std::collections::BTreeSet<_>>()
+            .into_iter()
+            .map(|hash| {
+                if !members.contains(&hash) {
+                    return Err(StagingError::StagingCorrupt(
+                        "history recipe is not owned by publication intent".to_owned(),
+                    ));
+                }
+                guard.verified_local_recipe(&hash.into())?.ok_or_else(|| {
+                    StagingError::StagingCorrupt(
+                        "publication history recipe is not verified".to_owned(),
+                    )
+                })
+            })
+            .collect::<Result<Vec<_>>>()?
+    };
+    // Commit every history owner before rollback can remove native leases.
+    // An interruption leaves the original intent and its remaining payloads owned.
+    for recipe in recipes {
+        if cancel.is_cancelled() {
+            return Err(StagingError::Cancelled);
+        }
+        let owner = PathBuf::from(".crab/history").join(recipe.file_hash().hex());
+        publish_verified_recipe_owner(root, index, &owner, &recipe, None)?;
+    }
+    Ok(())
+}
+
 #[cfg(unix)]
 fn staging_path_bytes(path: &Path) -> Vec<u8> {
     use std::os::unix::ffi::OsStrExt;
@@ -2707,6 +2800,16 @@ impl StagingArea {
         unresolved_publication_intents(&self.index)
     }
 
+    /// Preserve caller-attested committed recipe roots before rolling back a native intent.
+    pub fn preserve_publication_history(
+        &self,
+        intent_id: &PublicationIntentId,
+        recipe_hashes: &[MerkleHash],
+        cancel: &tokio_util::sync::CancellationToken,
+    ) -> Result<()> {
+        preserve_publication_history(&self.root, &self.index, intent_id, recipe_hashes, cancel)
+    }
+
     /// Atomically remove an unresolved intent and every open batch it owns.
     pub fn rollback_publication_intent(
         &self,
@@ -2726,31 +2829,13 @@ impl StagingArea {
         path: &Path,
         recipe: &crate::recipe::FileRecipe,
     ) -> Result<StagingBatchId> {
-        self.publish_verified_recipe_owner(path, recipe, Some(&path.to_string_lossy()))
-    }
-
-    fn publish_verified_recipe_owner(
-        &self,
-        owner: &Path,
-        recipe: &crate::recipe::FileRecipe,
-        display_path: Option<&str>,
-    ) -> Result<StagingBatchId> {
-        let batch_id = self.create_batch()?;
-        let file_hash = recipe.file_hash();
-        let publish = (|| -> Result<()> {
-            self.pre_register_file(&file_hash, recipe.file_size())?;
-            self.record_verified_recipe_lease(&batch_id, owner, recipe)?;
-            if let Some(display_path) = display_path {
-                self.record_file_path(&file_hash, display_path)?;
-            }
-            self.mark_batch_published(&batch_id)
-        })();
-        if let Err(error) = publish {
-            let _ = self.rollback_batch(&batch_id);
-            let _ = self.retire_file_if_unleased(&file_hash);
-            return Err(error);
-        }
-        Ok(batch_id)
+        publish_verified_recipe_owner(
+            &self.root,
+            &self.index,
+            path,
+            recipe,
+            Some(&path.to_string_lossy()),
+        )
     }
 
     /// Publish one immutable recipe needed by not-yet-pushed Git history.
@@ -2763,7 +2848,7 @@ impl StagingArea {
         recipe: &crate::recipe::FileRecipe,
     ) -> Result<StagingBatchId> {
         let owner = PathBuf::from(".crab/history").join(recipe.file_hash().hex());
-        self.publish_verified_recipe_owner(&owner, recipe, None)
+        publish_verified_recipe_owner(&self.root, &self.index, &owner, recipe, None)
     }
 
     /// Preserve a locally published recipe referenced by not-yet-pushed Git history.
@@ -4134,6 +4219,16 @@ impl StagingAreaReadOnly {
         Ok(())
     }
 
+    /// Preserve caller-attested committed recipe roots before rolling back a native intent.
+    pub fn preserve_publication_history(
+        &self,
+        intent_id: &PublicationIntentId,
+        recipe_hashes: &[MerkleHash],
+        cancel: &tokio_util::sync::CancellationToken,
+    ) -> Result<()> {
+        preserve_publication_history(&self.root, &self.index, intent_id, recipe_hashes, cancel)
+    }
+
     /// Atomically remove an unresolved intent and every open batch it owns.
     pub async fn rollback_publication_intent(
         &self,
@@ -4165,6 +4260,10 @@ impl StagingAreaReadOnly {
 
     /// Retire the exact recipe ownership pinned by one committed push snapshot.
     pub async fn retire_push_snapshot(&self, push_id: &str) -> Result<Vec<RetireStats>> {
+        self.retire_push_snapshot_now(push_id)
+    }
+
+    fn retire_push_snapshot_now(&self, push_id: &str) -> Result<Vec<RetireStats>> {
         let unleased = lock_index(&self.index)?.retire_push_snapshot(push_id)?;
         remove_indexed_files(&self.root, &self.index, &unleased)
     }
diff --git a/crates/crab-staging/src/push_snapshot.rs b/crates/crab-staging/src/push_snapshot.rs
new file mode 100644
index 000000000..2437a4634
--- /dev/null
+++ b/crates/crab-staging/src/push_snapshot.rs
@@ -0,0 +1,70 @@
+use std::sync::Arc;
+
+use crate::{Result, StagingAreaReadOnly, recipe::FileRecipe};
+
+/// Own one push reader's marker and exact immutable recipe snapshot.
+#[must_use = "retain the snapshot through remote publication"]
+pub struct StagingPushSnapshot {
+    staging: Arc<StagingAreaReadOnly>,
+    push_id: String,
+    committed: bool,
+}
+
+impl StagingPushSnapshot {
+    /// Fence retirement before selecting recipes for a uniquely identified push.
+    ///
+    /// The caller must supply a unique push ID.
+    /// Failure or drop preserves their original leases; another reader can defer cleanup.
+    pub fn begin(staging: Arc<StagingAreaReadOnly>, push_id: String) -> Result<Self> {
+        let snapshot = Self {
+            staging,
+            push_id,
+            committed: false,
+        };
+        snapshot.staging.mark_push_inflight(&snapshot.push_id)?;
+        Ok(snapshot)
+    }
+
+    /// Pin published/delegated recipes selected from verified Git pointers.
+    pub fn capture(&self, recipes: &[FileRecipe]) -> Result<()> {
+        self.staging.create_push_snapshot(&self.push_id, recipes)
+    }
+
+    /// Record confirmed remote publication and retire only captured local owners.
+    ///
+    /// The caller must establish remote commit before calling this method. Contention
+    /// leaves a committed journal for exclusive maintenance; cleanup errors retain it.
+    pub fn commit(mut self) -> Result<()> {
+        // Remote commit is already authoritative. Even if the local transition
+        // fails, Drop must retain its journal rather than treat it as a failed push.
+        self.committed = true;
+        self.staging.commit_push_snapshot(&self.push_id)?;
+        let Some(_retirement) = self.staging.begin_retirement(&self.push_id)? else {
+            return Ok(());
+        };
+        self.staging.retire_push_snapshot_now(&self.push_id)?;
+        self.staging.sweep_orphans()?;
+        self.staging.remove_push_snapshot(&self.push_id)
+    }
+}
+
+impl Drop for StagingPushSnapshot {
+    fn drop(&mut self) {
+        if !self.committed {
+            let discard = (|| -> Result<()> {
+                // A sibling may still read a version this snapshot alone pins.
+                // Exclusive maintenance consumes the journal if cleanup is deferred.
+                if let Some(_retirement) = self.staging.begin_retirement(&self.push_id)? {
+                    self.staging.discard_open_push_snapshot_now(&self.push_id)?;
+                }
+                Ok(())
+            })();
+            if let Err(error) = discard {
+                tracing::warn!(push_id = %self.push_id, %error, "staging snapshot cleanup needs maintenance");
+            }
+        }
+        if let Err(error) = self.staging.clear_push_inflight(&self.push_id) {
+            tracing::warn!(push_id = %self.push_id, %error, "staging push marker cleanup failed");
+        }
+    }
+}
diff --git a/crates/crab-staging/tests/publication_history_recovery.rs b/crates/crab-staging/tests/publication_history_recovery.rs
new file mode 100644
index 000000000..7afe7de79
--- /dev/null
+++ b/crates/crab-staging/tests/publication_history_recovery.rs
@@ -0,0 +1,128 @@
+//! Committed history ownership must outlive rollback of its native index intent.
+
+#![expect(clippy::unwrap_used, reason = "test assertions")]
+
+use std::path::Path;
+use std::sync::Arc;
+
+use crab_staging::stream::{StreamStageProgress, StreamStageXorbBuilder, stage_reader_streaming};
+use crab_staging::{
+    PublicationIntentEntry, StagingArea, StagingAreaReadOnly, StagingError, StagingPushSnapshot,
+};
+use crab_xet::hash::MerkleHash;
+use tokio_util::sync::CancellationToken;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn committed_history_preserves_only_selected_intent_recipes_through_rollback() {
+    let temp = tempfile::tempdir().unwrap();
+    let staging = StagingArea::open(temp.path().to_owned()).await.unwrap();
+    let mut staged = Vec::new();
+    for (path, byte) in [("committed", 0x71), ("uncommitted", 0x72)] {
+        staged.push(
+            stage_reader_streaming(
+                &mut std::io::Cursor::new(vec![byte; 2 * 1024 * 1024]),
+                Path::new(path),
+                &staging,
+                StreamStageProgress {
+                    xorb_builder: Some(StreamStageXorbBuilder::new(
+                        1,
+                        crab_xet::xorb::builder::XorbBuilder::new,
+                    )),
+                    ..Default::default()
+                },
+                &CancellationToken::new(),
+            )
+            .await
+            .unwrap(),
+        );
+    }
+    let entries = staged
+        .iter()
+        .enumerate()
+        .map(|(index, staged)| PublicationIntentEntry {
+            batch_id: staged.batch_id.clone(),
+            path: if index == 0 {
+                "committed"
+            } else {
+                "uncommitted"
+            }
+            .into(),
+            expected_pointer_oid: format!("{}", index + 1).repeat(40),
+            previous_index_state: "0".repeat(64),
+        })
+        .collect::<Vec<_>>();
+    let intent = staging.create_publication_intent(&entries).unwrap();
+    let recipe_hash = MerkleHash::from(staged[0].recipe.hash());
+    let invalid = staging.preserve_publication_history(
+        &intent,
+        &[recipe_hash, MerkleHash::from([0xff; 32])],
+        &CancellationToken::new(),
+    );
+    assert!(invalid.is_err() && staging.lifecycle_health().unwrap().path_heads == 0);
+    let cancelled = CancellationToken::new();
+    cancelled.cancel();
+    assert!(matches!(
+        staging.preserve_publication_history(&intent, &[recipe_hash], &cancelled),
+        Err(StagingError::Cancelled)
+    ));
+    for _ in 0..2 {
+        staging
+            .preserve_publication_history(&intent, &[recipe_hash], &CancellationToken::new())
+            .unwrap();
+    }
+    let health = staging.lifecycle_health().unwrap();
+    assert!(
+        health.path_heads == 1
+            && health.path_leases == 3
+            && health.payloads == 0
+            && health.current_head_prepared_bytes > 0,
+        "{health:?}"
+    );
+    staging.rollback_publication_intent(&intent).unwrap();
+    assert!(
+        staging
+            .published_recipe_for_file(&staged[0].recipe.file_hash())
+            .unwrap()
+            == Some(staged[0].recipe.clone())
+            && staging
+                .chunks_for_file(&staged[0].recipe.file_hash())
+                .unwrap()
+                .len()
+                == 16
+            && staging
+                .chunks_for_file(&staged[1].recipe.file_hash())
+                .unwrap()
+                .is_empty()
+    );
+    staging.close().await.unwrap();
+    let reader = Arc::new(
+        StagingAreaReadOnly::open(temp.path().to_owned())
+            .await
+            .unwrap(),
+    );
+    reader
+        .verify_file_reconstruction(&staged[0].recipe.file_hash(), staged[0].size)
+        .await
+        .unwrap();
+    let prepared = reader.prepared_xorbs_for_recipes(&[recipe_hash]).unwrap();
+    let bodies = prepared
+        .values()
+        .flatten()
+        .map(|xorb| {
+            crab_staging::push_plan::prepared_xorb_path(
+                temp.path(),
+                &MerkleHash::from_hex(&xorb.hash).unwrap(),
+            )
+        })
+        .collect::<Vec<_>>();
+    assert!(!bodies.is_empty() && bodies.iter().all(|path| path.is_file()));
+    let snapshot =
+        StagingPushSnapshot::begin(Arc::clone(&reader), "history-push".to_owned()).unwrap();
+    snapshot
+        .capture(std::slice::from_ref(&staged[0].recipe))
+        .unwrap();
+    snapshot.commit().unwrap();
+    let health = reader.lifecycle_health().unwrap();
+    assert!(health.path_heads == 0 && health.path_leases == 0 && health.payloads == 0);
+    assert!(bodies.iter().all(|path| !path.try_exists().unwrap()));
+}
diff --git a/crates/crab-staging/tests/push_snapshot_guard.rs b/crates/crab-staging/tests/push_snapshot_guard.rs
new file mode 100644
index 000000000..909622b41
--- /dev/null
+++ b/crates/crab-staging/tests/push_snapshot_guard.rs
@@ -0,0 +1,174 @@
+//! Reader fences and exact lease cleanup across push completion and cancellation.
+
+#![expect(clippy::unwrap_used, reason = "test assertions")]
+
+use std::path::Path;
+use std::sync::Arc;
+
+use crab_staging::stream::{StreamStageProgress, stage_reader_streaming};
+use crab_staging::{StagingArea, StagingAreaReadOnly, StagingPushSnapshot};
+use tokio_util::sync::CancellationToken;
+
+async fn fixture(
+    root: &Path,
+    native_owner: bool,
+) -> (
+    Arc<StagingAreaReadOnly>,
+    crab_staging::stream::StreamStageResult,
+    Option<crab_staging::StagingBatchId>,
+) {
+    let staging = StagingArea::open(root.to_owned()).await.unwrap();
+    let staged = stage_reader_streaming(
+        &mut std::io::Cursor::new(b"shared fixture payload"),
+        Path::new("model.bin"),
+        &staging,
+        StreamStageProgress::default(),
+        &CancellationToken::new(),
+    )
+    .await
+    .unwrap();
+    staging
+        .delegate_publication(&[(&staged.batch_id, Path::new("model.bin"))])
+        .unwrap();
+    let native = if native_owner {
+        let batch = staging.create_batch().unwrap();
+        staging
+            .record_verified_recipe_lease(&batch, Path::new("native.bin"), &staged.recipe)
+            .unwrap();
+        Some(batch)
+    } else {
+        None
+    };
+    staging.close().await.unwrap();
+    (
+        Arc::new(StagingAreaReadOnly::open(root.to_owned()).await.unwrap()),
+        staged,
+        native,
+    )
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn dropped_snapshot_keeps_delegated_payload_and_clears_its_reader() {
+    let temp = tempfile::tempdir().unwrap();
+    let (staging, staged, _) = fixture(temp.path(), false).await;
+    let snapshot = StagingPushSnapshot::begin(Arc::clone(&staging), "failed".to_owned()).unwrap();
+    snapshot
+        .capture(std::slice::from_ref(&staged.recipe))
+        .unwrap();
+    drop(snapshot);
+    let health = staging.lifecycle_health().unwrap();
+    assert!(
+        health.open_push_snapshots == 0
+            && health.path_leases == 1
+            && health.payloads == 1
+            && staging.list_inflight().unwrap().is_empty()
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn committed_snapshot_preserves_active_native_owner_of_shared_payload() {
+    let temp = tempfile::tempdir().unwrap();
+    let (staging, staged, native) = fixture(temp.path(), true).await;
+    let snapshot =
+        StagingPushSnapshot::begin(Arc::clone(&staging), "committed".to_owned()).unwrap();
+    snapshot
+        .capture(std::slice::from_ref(&staged.recipe))
+        .unwrap();
+    snapshot.commit().unwrap();
+    let health = staging.lifecycle_health().unwrap();
+    assert!(
+        health.path_leases == 1
+            && health.payloads == 1
+            && health.open_push_snapshots == 0
+            && health.committed_push_snapshots == 0
+            && staging.list_inflight().unwrap().is_empty()
+    );
+    drop(staging);
+    let writer = StagingArea::open(temp.path().to_owned()).await.unwrap();
+    writer.rollback_batch(&native.unwrap()).unwrap();
+    assert!(
+        writer
+            .chunks_for_file(&staged.recipe.file_hash())
+            .unwrap()
+            .is_empty()
+    );
+    writer.close().await.unwrap();
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn sibling_reader_defers_retirement_until_exclusive_maintenance() {
+    let temp = tempfile::tempdir().unwrap();
+    let (staging, staged, _) = fixture(temp.path(), false).await;
+    let first = StagingPushSnapshot::begin(Arc::clone(&staging), "first".to_owned()).unwrap();
+    first.capture(std::slice::from_ref(&staged.recipe)).unwrap();
+    let second = StagingPushSnapshot::begin(Arc::clone(&staging), "second".to_owned()).unwrap();
+    second
+        .capture(std::slice::from_ref(&staged.recipe))
+        .unwrap();
+    first.commit().unwrap();
+    let health = staging.lifecycle_health().unwrap();
+    assert!(
+        health.committed_push_snapshots == 1
+            && health.open_push_snapshots == 1
+            && health.payloads == 1
+    );
+    drop(second);
+    drop(staging);
+    let writer = StagingArea::open(temp.path().to_owned()).await.unwrap();
+    writer.clean().unwrap();
+    let health = writer.lifecycle_health().unwrap();
+    assert!(
+        health.path_leases == 0
+            && health.payloads == 0
+            && health.open_push_snapshots == 0
+            && health.committed_push_snapshots == 0
+            && writer.list_inflight().unwrap().is_empty()
+    );
+    writer.close().await.unwrap();
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn task_abort_drops_snapshot_without_retiring_unpublished_lease() {
+    let temp = tempfile::tempdir().unwrap();
+    let (staging, staged, _) = fixture(temp.path(), false).await;
+    let reader = Arc::clone(&staging);
+    let (ready, received) = tokio::sync::oneshot::channel();
+    let task = tokio::spawn(async move {
+        let snapshot = StagingPushSnapshot::begin(reader, "aborted".to_owned()).unwrap();
+        snapshot
+            .capture(std::slice::from_ref(&staged.recipe))
+            .unwrap();
+        ready.send(()).unwrap();
+        std::future::pending::<()>().await;
+        drop(snapshot);
+    });
+    received.await.unwrap();
+    task.abort();
+    assert!(task.await.unwrap_err().is_cancelled());
+    let health = staging.lifecycle_health().unwrap();
+    assert!(
+        health.path_leases == 1
+            && health.open_push_snapshots == 0
+            && health.payloads == 1
+            && staging.list_inflight().unwrap().is_empty()
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn local_commit_failure_releases_marker_and_preserves_original_lease() {
+    let temp = tempfile::tempdir().unwrap();
+    let (staging, staged, _) = fixture(temp.path(), false).await;
+    let snapshot =
+        StagingPushSnapshot::begin(Arc::clone(&staging), "missing-journal".to_owned()).unwrap();
+    snapshot
+        .capture(std::slice::from_ref(&staged.recipe))
+        .unwrap();
+    staging.remove_push_snapshot("missing-journal").unwrap();
+    assert!(snapshot.commit().is_err());
+    let health = staging.lifecycle_health().unwrap();
+    assert!(
+        health.path_leases == 1
+            && health.payloads == 1
+            && staging.list_inflight().unwrap().is_empty()
+    );
+}