feat(pre-push-safety-net): materialize bounded isolated snapshots

.kiro/specs/pre-push-safety-net/brief.md

@@ -1,6 +1,6 @@
 # Brief: pre-push-safety-net
 
-Status: requirements, design and tasks approved on 2026-09-12; group 1 implementation verified complete; groups 2–8 pending. Language: English.
+Status: requirements, design and tasks approved on 2026-09-12; group 1 and task 2.1 verified complete; group 2 implementation underway. Language: English.
 
 ## Problem
 
@@ -10,8 +10,9 @@ maintainer explicitly rejects noisy advisory output as well as noisy blocking ou
 
 ## Current State
 
-The scan driver and Joern arbiters exist, with demonstrated PHP/Python cases. There is no pre-push
-snapshot resolver, change context for the ranker, persistent graph cache, or end-to-end deadline.
+The scan driver and Joern arbiters exist, with demonstrated PHP/Python cases. Immutable push-tip
+resolution is implemented and verified. Isolated source materialization and change context are
+under implementation; persistent graph caching and an end-to-end deadline remain pending.
 The driver builds a whole-root graph before cutting the region budget. The latest PMPro ranking
 experiment alone spends about 130 seconds building and 296 seconds extracting evidence. A manually
 selected two-file slice finding a CVE does not demonstrate automatic incremental analysis.

.kiro/specs/pre-push-safety-net/tasks.md

@@ -63,12 +63,13 @@ the hook and capability admission are not yet implemented. See `implementation-g
   - _Boundary: Snapshot Adapter_
   - _Requirements: 1.1, 1.3, 1.4, 7.1_
   - _Verified 2026-09-12:_ independent review APPROVED after Unicode-ref repair; 997 tests passed, 9 skipped; focused 76 passed and touched Ruff/mypy passed. Real SHA-1/SHA-256 fixtures cover exact comparisons, tags, deletion, missing/ambiguous bases, replacement/graft suppression and blocked promisor fetch. Parent read 2,547 pinned source bytes and verified real-repository comparison/ref identity with dirty state and index preserved. ASCII protocol framing retains valid Unicode ref names.
-- [ ] 2.2 Materialize isolated object snapshots
+- [x] 2.2 Materialize isolated object snapshots
   - Read immutable tracked blobs into bounded scratch storage using machine-readable paths; do not checkout into the live tree or run filters, project scripts or network requests.
   - Preserve path identity and report symlinks, gitlinks, unavailable LFS blobs and unsupported content; never follow source outside the snapshot.
   - Done when non-HEAD pushes from a dirty checkout read only the intended object bytes and leave index, files and refs byte-identical after success, cancellation and failure.
   - _Boundary: Snapshot Adapter_
   - _Requirements: 1.1, 1.2, 1.3, 2.3, 7.1, 7.3_
+  - _Verified 2026-09-12:_ independent review APPROVED; 1,037 tests passed, 9 skipped; focused 116 passed; touched Ruff/mypy passed. Real SHA-1/SHA-256 fixtures verify binary-safe paths, exact blobs, no filters, explicit symlink/gitlink/LFS limits, cancellation and unchanged live state. Independent hanging-child cleanup took 0.202 s. Parent reopened and checked all 1,157 materialized repository files (8,737,198 bytes), with scratch removed and live state preserved; evidence `benchmarks/measurements/2026-09-12-snapshot-materialization.json` records a 5.783 s preparation-only lab run, not end-to-end hook latency.
 - [ ] 2.3 Attach changed spans and deletion/rename context
   - Compare recorded base/head objects using NUL-safe path handling; retain removed spans, renamed identities and declaration/config changes.
   - Supply change context to the shared ranker contract rather than selecting an independent changed-files-only scope.

benchmarks/measurements/2026-09-12-snapshot-materialization.json

@@ -0,0 +1,20 @@
+{
+  "commit": "05da202d677622b56a8d1a2d767722015b84b336",
+  "base": "14a2a60e52bfaa08296da5b0bbe1043bece207fa",
+  "source_probe_bytes": 2547,
+  "files": 1157,
+  "bytes_read": 8737198,
+  "tree_complete": true,
+  "materialization_complete": true,
+  "boundaries": [],
+  "elapsed_seconds": 5.783,
+  "scratch_removed": true,
+  "live_state_preserved": true,
+  "budget_seconds": 120,
+  "purpose": "Snapshot correctness lab probe; not end-to-end hook latency",
+  "implementation_sha256": {
+    "src/openultrasast/push/snapshot.py": "fa0a77edc284f65430f8caeb7d5ebd949178b135573f28d69de017c7535ea2af",
+    "src/openultrasast/push/contracts.py": "5687fe8f7f7e3b6d40b76252e390e8236a653a28bd07e9e985bfaf6864916431"
+  },
+  "verification": "Every materialized file was reopened and checked against manifest byte length and SHA256; pyproject.toml also compared with Git blob bytes."
+}

src/openultrasast/push/contracts.py

@@ -5,6 +5,7 @@ No default result status can turn an unexecuted contract into a complete negativ
 """
 from __future__ import annotations
 
+import re
 from dataclasses import dataclass
 from typing import Literal
 
@@ -104,3 +105,72 @@ class PushResolution(Contract):
     updates: tuple[UpdateResolution, ...]
     comparisons: tuple[PushComparison, ...]
     targets: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class SnapshotFile(Contract):
+    # Git paths are bytes; hex survives JSON and whitespace-only/non-UTF8 names.
+    path_hex: str
+    blob_oid: str
+    mode: str
+    size_bytes: int
+    sha256: str
+
+    def __post_init__(self) -> None:
+        super().__post_init__()
+        _validate_path_hex(self.path_hex)
+        if self.size_bytes < 0 or self.mode not in ("100644", "100755"):
+            raise ValueError("invalid materialized file size or mode")
+        if re.fullmatch(r"[0-9a-f]{40}|[0-9a-f]{64}", self.blob_oid) is None:
+            raise ValueError("invalid blob identity")
+        if re.fullmatch(r"[0-9a-f]{64}", self.sha256) is None:
+            raise ValueError("invalid content digest")
+
+    @property
+    def path_bytes(self) -> bytes:
+        return bytes.fromhex(self.path_hex)
+
+
+@dataclass(frozen=True)
+class SnapshotBoundary(Contract):
+    path_hex: str | None
+    reason: str
+
+    def __post_init__(self) -> None:
+        super().__post_init__()
+        if self.path_hex is not None:
+            _validate_path_hex(self.path_hex)
+
+
+def _validate_path_hex(value: str) -> None:
+    if re.fullmatch(r"(?:[0-9a-f]{2})+", value) is None or b"\0" in bytes.fromhex(value):
+        raise ValueError("path must be nonempty NUL-free bytes encoded as lowercase hex")
+
+
+@dataclass(frozen=True)
+class SnapshotManifest(Contract):
+    commit_oid: str
+    object_format: str
+    files: tuple[SnapshotFile, ...]
+    boundaries: tuple[SnapshotBoundary, ...]
+    tree_complete: bool
+    bytes_read: int
+
+    def __post_init__(self) -> None:
+        super().__post_init__()
+        width = {"sha1": 40, "sha256": 64}.get(self.object_format)
+        if width is None or re.fullmatch(r"[0-9a-f]{" + str(width) + r"}", self.commit_oid) is None:
+            raise ValueError("invalid snapshot identity")
+        if any(len(item.blob_oid) != width for item in self.files):
+            raise ValueError("blob identity does not match object format")
+        if len({item.path_hex for item in self.files}) != len(self.files):
+            raise ValueError("snapshot paths must be unique")
+        if self.bytes_read < sum(item.size_bytes for item in self.files):
+            raise ValueError("bytes read must cover every materialized blob")
+        if not self.tree_complete and not self.boundaries:
+            raise ValueError("incomplete tree requires an explicit boundary")
+
+    @property
+    def complete(self) -> bool:
+        """Materialization completeness only, never security coverage."""
+        return self.tree_complete and not self.boundaries

src/openultrasast/push/snapshot.py

@@ -1,17 +1,34 @@
 """Resolve supplied push tips using local Git objects, never the live checkout.
 
 No fetch, replacement objects, checkout, filters or project commands are used.
-Materialization and changed spans are separate subsequent snapshot-adapter tasks.
+Snapshots contain exact tracked blobs; changed spans are a subsequent adapter task.
 """
 from __future__ import annotations
 
+import hashlib
 import os
 import re
+import selectors
+import signal
+import stat
 import subprocess
-from dataclasses import replace
+import tempfile
+import time
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager, suppress
+from dataclasses import dataclass, replace
 from pathlib import Path
 
-from openultrasast.push.contracts import PushComparison, PushResolution, PushUpdate, UpdateResolution
+from openultrasast.model.contracts import ExecutionBudget
+from openultrasast.push.contracts import (
+    PushComparison,
+    PushResolution,
+    PushUpdate,
+    SnapshotBoundary,
+    SnapshotFile,
+    SnapshotManifest,
+    UpdateResolution,
+)
 
 
 class SnapshotInputError(ValueError):
@@ -31,6 +48,18 @@ class SnapshotAdapter:
         # Ask the installed Git for its object ID width rather than assuming SHA-1.
         self._oid_width = len(self._required("hash-object", "--stdin", input=""))
         self._zero = "0" * self._oid_width
+        self._git_dir = self._repository_path("--absolute-git-dir")
+        self._common_dir = self._repository_path("--path-format=absolute", "--git-common-dir")
+        try:
+            self._worktree = self._repository_path("--show-toplevel")
+        except SnapshotInputError:
+            self._worktree = self.repository  # Bare repositories have no working tree.
+
+    def _repository_path(self, *args: str) -> Path:
+        # Binary framing also matters for the repository's own whitespace/non-UTF8 name.
+        raw = self._bounded_git(("rev-parse", *args), 65536,
+                                ExecutionBudget(time.monotonic() + 10.0, 2.0), None, [])
+        return Path(os.fsdecode(raw.removesuffix(b"\n"))).resolve()
 
     def _git(self, *args: str, input: str | None = None) -> subprocess.CompletedProcess[str]:
         try:
@@ -149,3 +178,198 @@ class SnapshotAdapter:
             else:
                 comparisons.append(PushComparison(*key, refs=(update.remote_ref,)))
         return PushResolution(self.object_format, tuple(records), tuple(comparisons), tuple(targets))
+
+    @contextmanager
+    def materialize(
+        self, commit_oid: str, *, budget: ExecutionBudget | None = None,
+        limits: SnapshotLimits | None = None, scratch_parent: Path | None = None,
+        cancelled: Callable[[], bool] | None = None,
+    ) -> Iterator[MaterializedSnapshot]:
+        """Own one disposable tracked-object tree, with no checkout/filter execution.
+
+        Partial manifests are usable only for diagnostics/explicit partial analysis.
+        Scratch is removed on normal return and exceptions, including cancellation.
+        """
+        budget = budget or ExecutionBudget(time.monotonic() + 30.0, 2.0)
+        limits = limits or SnapshotLimits()
+        if re.fullmatch(r"[0-9a-f]{" + str(self._oid_width) + r"}", commit_oid) is None:
+            raise SnapshotInputError("snapshot requires a full immutable commit OID")
+        parent = (scratch_parent or Path(tempfile.gettempdir())).resolve()
+        if any(parent == root or root in parent.parents for root in (self.repository, self._worktree, self._git_dir, self._common_dir)):
+            raise SnapshotInputError("scratch parent must be outside the checkout and Git metadata")
+        _check_budget(budget, cancelled)
+        cleanup_deadline: list[float] = []
+        root = Path(tempfile.mkdtemp(prefix="ousast-snapshot-", dir=parent))
+        files: list[SnapshotFile] = []
+        boundaries: list[SnapshotBoundary] = []
+        total = 0
+        tree_complete = False
+        try:
+            try:
+                if self._bounded_git(("cat-file", "-t", commit_oid), 64, budget, cancelled, cleanup_deadline) != b"commit\n":
+                    raise SnapshotInputError("unsupported_snapshot_object")
+                listing = self._bounded_git(("ls-tree", "-r", "-z", "--full-tree", commit_oid),
+                                            limits.max_tree_bytes, budget, cancelled, cleanup_deadline)
+                records = listing.split(b"\0")
+                if records[-1] != b"":
+                    raise SnapshotInputError("truncated_tree_metadata")
+                records.pop()
+                if len(records) > limits.max_files:
+                    raise SnapshotInputError("file_count_limit")
+                tree_complete = True
+                for record in records:
+                    _check_budget(budget, cancelled)
+                    metadata, path = record.split(b"\t", 1)
+                    mode, kind, oid = metadata.split(b" ")
+                    path_hex = path.hex()
+                    components = path.split(b"/")
+                    if (not path or path.startswith(b"/") or any(part in (b"", b".", b"..") for part in components)
+                            or len(path) > limits.max_path_bytes or len(components) > limits.max_path_depth
+                            or any(part.lower() == b".git" for part in components)):
+                        boundaries.append(SnapshotBoundary(path_hex or None, "unsupported_path"))
+                        continue
+                    if mode == b"120000":
+                        boundaries.append(SnapshotBoundary(path_hex, "symlink_not_materialized"))
+                        continue
+                    if mode == b"160000":
+                        boundaries.append(SnapshotBoundary(path_hex, "gitlink_not_materialized"))
+                        continue
+                    if kind != b"blob" or mode not in (b"100644", b"100755"):
+                        boundaries.append(SnapshotBoundary(path_hex, "unsupported_tree_entry"))
+                        continue
+                    blob_oid = oid.decode("ascii")
+                    try:
+                        size = int(self._bounded_git(("cat-file", "-s", blob_oid), 64, budget, cancelled, cleanup_deadline))
+                        if size < 0 or size > limits.max_blob_bytes or size > limits.max_total_bytes - total:
+                            raise SnapshotInputError("source_byte_limit")
+                        data = self._bounded_git(("cat-file", "blob", blob_oid), size, budget, cancelled, cleanup_deadline)
+                        if len(data) != size:
+                            raise SnapshotInputError("blob_size_mismatch")
+                        digest = hashlib.new(self.object_format, b"blob " + str(size).encode() + b"\0" + data).hexdigest()
+                        if digest != blob_oid:
+                            raise SnapshotInputError("blob_identity_mismatch")
+                        total += size
+                        if data.startswith(b"version https://git-lfs.github.com/spec/v1\n"):
+                            raise SnapshotInputError("lfs_content_unavailable")
+                        target = root / os.fsdecode(path)
+                        target.parent.mkdir(parents=True, exist_ok=True)
+                        with target.open("xb") as output:
+                            output.write(data)
+                        target.chmod(0o555 if mode == b"100755" else 0o444)
+                        files.append(SnapshotFile(path_hex, blob_oid, mode.decode("ascii"), size,
+                                                  hashlib.sha256(data).hexdigest()))
+                    except SnapshotInputError as error:
+                        boundaries.append(SnapshotBoundary(path_hex, str(error)))
+                        if str(error) in ("deadline_exhausted", "cancelled", "git_output_limit", "git_cleanup_deadline_exhausted"):
+                            break
+            except (SnapshotInputError, OSError, ValueError) as error:
+                boundaries.append(SnapshotBoundary(None, str(error) or type(error).__name__))
+            yield MaterializedSnapshot(root, SnapshotManifest(commit_oid, self.object_format, tuple(files),
+                                                             tuple(boundaries), tree_complete, total))
+        finally:
+            _remove_scratch(root, _cleanup_deadline(budget, cleanup_deadline))
+
+    def _bounded_git(self, args: tuple[str, ...], limit: int, budget: ExecutionBudget,
+                     cancelled: Callable[[], bool] | None, cleanup_deadline: list[float]) -> bytes:
+        _check_budget(budget, cancelled)
+        command = ["git", "-c", "protocol.allow=never", "-c", "core.hooksPath=" + os.devnull,
+                   "-C", str(self.repository), *args]
+        try:
+            process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
+                                       stdin=subprocess.DEVNULL, env=self._env, start_new_session=True)
+        except OSError as error:
+            raise SnapshotInputError("git_launch_failed") from error
+        output = bytearray()
+        try:
+            assert process.stdout is not None
+            with selectors.DefaultSelector() as selector:
+                selector.register(process.stdout, selectors.EVENT_READ)
+                while selector.get_map():
+                    _check_budget(budget, cancelled)
+                    for key, _ in selector.select(min(0.05, max(0, budget.deadline_monotonic - time.monotonic()))):
+                        chunk = os.read(key.fd, min(65536, limit - len(output) + 1))
+                        if not chunk:
+                            selector.unregister(key.fileobj)
+                        else:
+                            output.extend(chunk)
+                            if len(output) > limit:
+                                raise SnapshotInputError("git_output_limit")
+            while process.poll() is None:
+                _check_budget(budget, cancelled)
+                time.sleep(0.001)
+            if process.returncode:
+                raise SnapshotInputError(f"local_git_{args[0]}_failed_exit_{process.returncode}")
+            return bytes(output)
+        finally:
+            # Kill the entire group even if its leader exited while a child held stdout.
+            with suppress(ProcessLookupError):
+                os.killpg(process.pid, signal.SIGKILL)
+            if process.poll() is None:
+                deadline = _cleanup_deadline(budget, cleanup_deadline)
+                try:
+                    process.wait(timeout=max(0, deadline - time.monotonic()))
+                except subprocess.TimeoutExpired as error:
+                    raise SnapshotInputError("git_cleanup_deadline_exhausted") from error
+            if process.stdout is not None:
+                process.stdout.close()
+
+
+@dataclass(frozen=True)
+class SnapshotLimits:
+    max_files: int = 10000
+    max_total_bytes: int = 128 * 1024 * 1024
+    max_blob_bytes: int = 8 * 1024 * 1024
+    max_tree_bytes: int = 4 * 1024 * 1024
+    max_path_bytes: int = 4095
+    max_path_depth: int = 64
+
+    def __post_init__(self) -> None:
+        for value in vars(self).values():
+            if type(value) is not int or value <= 0:
+                raise ValueError("snapshot limits must be positive integers")
+
+
+@dataclass(frozen=True)
+class MaterializedSnapshot:
+    root: Path
+    manifest: SnapshotManifest
+
+
+def _check_budget(budget: ExecutionBudget, cancelled: Callable[[], bool] | None) -> None:
+    if cancelled is not None and cancelled():
+        raise SnapshotInputError("cancelled")
+    if time.monotonic() >= budget.deadline_monotonic:
+        raise SnapshotInputError("deadline_exhausted")
+
+
+def _cleanup_deadline(budget: ExecutionBudget, recorded: list[float]) -> float:
+    if not recorded:
+        recorded.append(time.monotonic() + budget.cancellation_allowance_seconds)
+    return recorded[0]
+
+
+def _remove_scratch(root: Path, deadline: float) -> None:
+    # Directory-relative, no-follow operations avoid deleting through substituted links.
+    def remove_contents(descriptor: int) -> None:
+        for name in os.listdir(descriptor):
+            if time.monotonic() >= deadline:
+                raise SnapshotInputError("scratch_cleanup_deadline_exhausted: " + str(root))
+            info = os.stat(name, dir_fd=descriptor, follow_symlinks=False)
+            if stat.S_ISDIR(info.st_mode):
+                child = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=descriptor)
+                try:
+                    remove_contents(child)
+                finally:
+                    os.close(child)
+                os.rmdir(name, dir_fd=descriptor)
+            else:
+                os.unlink(name, dir_fd=descriptor)
+    if root.is_symlink():
+        root.unlink()
+        return
+    descriptor = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+    try:
+        remove_contents(descriptor)
+    finally:
+        os.close(descriptor)
+    root.rmdir()

tests/test_push_snapshot.py

@@ -205,3 +205,243 @@ def test_git_valid_unicode_ref_names_are_preserved(repo: Path, name: str) -> Non
     assert result.updates[0].update.remote_ref == ref
     assert result.comparisons[0].refs == (ref,)
     assert result.updates[0].disposition == "ready"
+
+
+def test_materialize_reads_non_head_objects_and_cleans(repo: Path, tmp_path: Path) -> None:
+    pushed = commit(repo, "<?php echo $_GET['x'];")
+    checked = commit(repo, "dirty baseline")
+    git(repo, "update-ref", "HEAD", checked)
+    (repo / "file.js").write_bytes(b"staged")
+    git(repo, "add", "file.js")
+    (repo / "file.js").write_bytes(b"dirty")
+    before = (repo / ".git/index").read_bytes()
+    with SnapshotAdapter(repo).materialize(pushed) as snapshot:
+        root = snapshot.root
+        assert (root / "file.js").read_bytes() == b"<?php echo $_GET['x'];"
+        assert snapshot.manifest.complete
+        assert snapshot.manifest.bytes_read == 22
+        assert snapshot.manifest.commit_oid == pushed
+    assert not root.exists()
+    assert (repo / ".git/index").read_bytes() == before
+    assert (repo / "file.js").read_bytes() == b"dirty"
+    assert git(repo, "rev-parse", "HEAD") == checked
+
+
+def binary_commit(repo: Path, entries: list[tuple[bytes, bytes, bytes]]) -> str:
+    records = []
+    for mode, path, data in entries:
+        if mode == b"160000":
+            oid = data
+            kind = b"commit"
+        else:
+            oid = subprocess.run(["git", "-C", str(repo), "hash-object", "-w", "--stdin"],
+                                 input=data, capture_output=True, check=True).stdout.rstrip(b"\n")
+            kind = b"blob"
+            assert int(git(repo, "cat-file", "-s", oid.decode())) == len(data)
+        records.append(mode + b" " + kind + b" " + oid + b"\t" + path + b"\0")
+    tree = subprocess.run(["git", "-C", str(repo), "mktree", "-z"], input=b"".join(records),
+                          capture_output=True, check=True).stdout.decode().strip()
+    return git(repo, "commit-tree", tree, input="binary fixture")
+
+
+def test_snapshot_exact_binary_paths_and_no_filters(repo: Path) -> None:
+    import os
+    entries = [(b"100644", path, data) for path, data in [
+        (b"handler.php", b"<?php echo $_GET['x'];\n"),
+        (b"api.js", b"app.post('/x', (req,res)=>res.send(req.body.x));\n"),
+        (b"newline\nname\t.js", b"\x00\xffbinary\r\n"), ("\u03b1.js".encode(), b"unicode"),
+        (b"\xff.js", b"invalid UTF8 name"), (b" ", b"space"),
+        (b".gitattributes", b"* filter=trap\n"),
+    ]]
+    oid = binary_commit(repo, entries)
+    sentinel = repo / "FILTER_RAN"
+    git(repo, "config", "filter.trap.smudge", "touch " + str(sentinel))
+    with SnapshotAdapter(repo).materialize(oid) as snapshot:
+        assert snapshot.manifest.complete
+        assert snapshot.manifest.bytes_read == sum(len(data) for _, _, data in entries)
+        assert {f.path_bytes for f in snapshot.manifest.files} == {path for _, path, _ in entries}
+        for _, path, data in entries:
+            assert (snapshot.root / os.fsdecode(path)).read_bytes() == data
+        assert type(snapshot.manifest).from_payload(snapshot.manifest.to_payload()) == snapshot.manifest
+    assert not sentinel.exists()
+
+
+def test_snapshot_omits_symlinks_gitlinks_and_lfs(repo: Path) -> None:
+    target = commit(repo, "submodule")
+    oid = binary_commit(repo, [
+        (b"120000", b"outside", b"/etc/passwd"),
+        (b"160000", b"submodule", target.encode()),
+        (b"100644", b"large.js", b"version https://git-lfs.github.com/spec/v1\noid sha256:" + b"0" * 64 + b"\nsize 3\n"),
+    ])
+    with SnapshotAdapter(repo).materialize(oid) as snapshot:
+        assert not snapshot.manifest.complete
+        assert snapshot.manifest.tree_complete
+        assert {b.reason for b in snapshot.manifest.boundaries} == {
+            "symlink_not_materialized", "gitlink_not_materialized", "lfs_content_unavailable"}
+        assert list(snapshot.root.iterdir()) == []
+
+
+@pytest.mark.parametrize("limit,reason", [("max_files", "file_count_limit"),
+    ("max_tree_bytes", "git_output_limit"), ("max_blob_bytes", "source_byte_limit"),
+    ("max_total_bytes", "source_byte_limit"), ("max_path_bytes", "unsupported_path")])
+def test_snapshot_limits_are_explicit(repo: Path, limit: str, reason: str) -> None:
+    from openultrasast.push.snapshot import SnapshotLimits
+    oid = binary_commit(repo, [(b"100644", b"one.js", b"123"), (b"100644", b"two.php", b"456")])
+    with SnapshotAdapter(repo).materialize(oid, limits=SnapshotLimits(**{limit: 1})) as snapshot:
+        root = snapshot.root
+        assert not snapshot.manifest.complete
+        assert reason in {b.reason for b in snapshot.manifest.boundaries}
+        assert snapshot.manifest.bytes_read <= 1 if limit == "max_total_bytes" else True
+    assert not root.exists()
+
+
+@pytest.mark.parametrize("failure", ["cancel", "deadline", "missing_blob", "consumer"])
+def test_snapshot_failure_preserves_live_state_and_cleans(repo: Path, failure: str) -> None:
+    import time
+
+    from openultrasast.model.contracts import ExecutionBudget
+    from openultrasast.push.snapshot import SnapshotInputError
+    oid = commit(repo, "object bytes")
+    git(repo, "update-ref", "HEAD", oid)
+    (repo / "file.js").write_bytes(b"staged")
+    git(repo, "add", "file.js")
+    (repo / "file.js").write_bytes(b"dirty")
+    hooks = repo / ".git/hooks/pre-push"
+    hooks.write_bytes(b"existing hook")
+    if failure == "missing_blob":
+        blob = git(repo, "rev-parse", oid + ":file.js")
+        (repo / ".git/objects" / blob[:2] / blob[2:]).unlink()
+    before = {p.relative_to(repo): p.read_bytes() for p in repo.rglob("*") if p.is_file()}
+    calls = 0
+    def cancel() -> bool:
+        nonlocal calls
+        calls += 1
+        return failure == "cancel" and calls >= 5
+    budget = ExecutionBudget(time.monotonic() + (-1 if failure == "deadline" else 10), 2.0)
+    root = None
+    try:
+        with SnapshotAdapter(repo).materialize(oid, budget=budget, cancelled=cancel) as snapshot:
+            root = snapshot.root
+            if failure == "consumer":
+                raise RuntimeError("consumer cancelled")
+            assert not snapshot.manifest.complete
+            assert snapshot.manifest.boundaries
+    except RuntimeError:
+        assert failure == "consumer"
+    except SnapshotInputError as error:
+        assert failure == "deadline" and str(error) == "deadline_exhausted"
+    assert (root is None and failure == "deadline") or (root is not None and not root.exists())
+    after = {p.relative_to(repo): p.read_bytes() for p in repo.rglob("*") if p.is_file()}
+    assert before == after
+
+
+def test_snapshot_rejects_live_scratch_parent(repo: Path) -> None:
+    from openultrasast.push.snapshot import SnapshotInputError
+    oid = commit(repo, "input")
+    for parent in (repo, repo / ".git", repo / ".git/objects"):
+        with pytest.raises(SnapshotInputError, match="outside"), SnapshotAdapter(repo).materialize(oid, scratch_parent=parent):
+            pytest.fail("unsafe scratch accepted")
+
+
+def test_snapshot_missing_tree_and_unsupported_object_are_not_clean(repo: Path) -> None:
+    oid = commit(repo, "input")
+    blob = git(repo, "rev-parse", oid + ":file.js")
+    with SnapshotAdapter(repo).materialize(blob) as snapshot:
+        assert not snapshot.manifest.complete
+        assert not snapshot.manifest.tree_complete
+    tree = git(repo, "rev-parse", oid + "^{tree}")
+    (repo / ".git/objects" / tree[:2] / tree[2:]).unlink()
+    with SnapshotAdapter(repo).materialize(oid) as snapshot:
+        assert not snapshot.manifest.complete
+        assert not snapshot.manifest.tree_complete
+        assert not snapshot.manifest.files
+
+
+def test_snapshot_refuses_oversize_blob_before_reading_it(repo: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    from openultrasast.push.snapshot import SnapshotLimits
+    oid = commit(repo, "large payload")
+    adapter = SnapshotAdapter(repo)
+    original = adapter._bounded_git
+    commands = []
+    def capture(args, *rest):
+        commands.append(args)
+        return original(args, *rest)
+    monkeypatch.setattr(adapter, "_bounded_git", capture)
+    with adapter.materialize(oid, limits=SnapshotLimits(max_blob_bytes=1)) as snapshot:
+        assert not snapshot.manifest.complete
+    assert not any(args[:2] == ("cat-file", "blob") for args in commands)
+
+
+def test_cleanup_does_not_follow_consumer_symlinks(repo: Path, tmp_path_factory: pytest.TempPathFactory) -> None:
+    outside = tmp_path_factory.mktemp("outside")
+    (outside / "keep").write_bytes(b"preserve")
+    oid = commit(repo, "source")
+    with SnapshotAdapter(repo).materialize(oid) as snapshot:
+        (snapshot.root / "external").symlink_to(outside, target_is_directory=True)
+    assert (outside / "keep").read_bytes() == b"preserve"
+    with SnapshotAdapter(repo).materialize(oid) as snapshot:
+        (snapshot.root / "file.js").unlink()
+        snapshot.root.rmdir()
+        snapshot.root.symlink_to(outside, target_is_directory=True)
+    assert (outside / "keep").read_bytes() == b"preserve"
+    assert not snapshot.root.exists()
+
+
+def test_snapshot_deadline_kills_slow_git_and_cleans(repo: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    import sys
+    import time
+
+    from openultrasast.model.contracts import ExecutionBudget
+    oid = commit(repo, "input")
+    adapter = SnapshotAdapter(repo)
+    real_popen = subprocess.Popen
+    processes = []
+    def slow_git(command, **kwargs):
+        process = real_popen([sys.executable, "-c", "import time; time.sleep(30)"], **kwargs)
+        processes.append(process)
+        return process
+    monkeypatch.setattr(subprocess, "Popen", slow_git)
+    start = time.monotonic()
+    with adapter.materialize(oid, budget=ExecutionBudget(start + 0.1, 1.0)) as snapshot:
+        assert {b.reason for b in snapshot.manifest.boundaries} == {"deadline_exhausted"}
+        assert not snapshot.manifest.complete
+    assert time.monotonic() - start < 1.5
+    assert processes and all(process.poll() is not None for process in processes)
+    assert not snapshot.root.exists()
+
+
+def test_snapshot_manifest_rejects_forged_census(repo: Path) -> None:
+    from dataclasses import replace
+    oid = commit(repo, "input")
+    with SnapshotAdapter(repo).materialize(oid) as snapshot:
+        manifest = snapshot.manifest
+        file = manifest.files[0]
+        for changes in ({"bytes_read": -1}, {"files": (file, file)}, {"tree_complete": False}):
+            with pytest.raises(ValueError):
+                replace(manifest, **changes)
+        for changes in ({"path_hex": "zz"}, {"size_bytes": -1}, {"sha256": "bad"}, {"path_hex": "00"}):
+            with pytest.raises(ValueError):
+                replace(file, **changes)
+
+
+def test_snapshot_subdirectory_adapter_cannot_scratch_in_live_parent(repo: Path) -> None:
+    from openultrasast.push.snapshot import SnapshotInputError
+    oid = commit(repo, "input")
+    nested = repo / "nested"
+    nested.mkdir()
+    with pytest.raises(SnapshotInputError, match="outside"), SnapshotAdapter(nested).materialize(oid, scratch_parent=repo):
+        pytest.fail("live parent accepted")
+
+
+def test_snapshot_write_failure_cleans_partial_scratch(repo: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    oid = commit(repo, "input")
+    original = Path.open
+    def broken(path, mode="r", *args, **kwargs):
+        if mode == "xb":
+            raise OSError("scratch_write_failed")
+        return original(path, mode, *args, **kwargs)
+    monkeypatch.setattr(Path, "open", broken)
+    with SnapshotAdapter(repo).materialize(oid) as snapshot:
+        assert not snapshot.manifest.complete
+        assert any(b.reason == "scratch_write_failed" for b in snapshot.manifest.boundaries)
+    assert not snapshot.root.exists()