fix(push): retain and retire capsule staging

.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

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.

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()

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,17 +119,26 @@ 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:
             add = runner.run_crab(repo, ["add", "--jsonl", "*.bin"], name=f"v{version} Crab add")
         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")