feat(pre-push-safety-net): publish and evict complete local artifacts safely

.kiro/specs/pre-push-safety-net/implementation-group-6.md

@@ -37,3 +37,35 @@ graphs cannot qualify. Query locations normalize to repository-relative paths.
 - CLAIM: Task 6.1 identity and lease behavior, not persistent runner reuse or rollout readiness.
 - EVIDENCE: tests/test_graph_reuse.py, 105 focused passing tests, full suite and real Joern control above; measurement JSON retained in benchmarks/measurements/2026-09-13-graph-lease-control.json.
 - GAPS: Warm push feasibility and independent capability eligibility remain unproven.
+
+## Task 6.2 brief and review
+
+Requirements 4.1, 4.3, 4.4 and 7.3; design: Cache and Engine Boundary. Private
+0700 cache roots, 0600 files, fixed lock stripes plus a publication lock, streamed payload
+digests, manifest receipts, atomic completed directory publication and size accounting
+(payload plus manifests) implement local storage. Incomplete data is never published as
+complete. A lookup holds a shared lease; eviction requires an exclusive lock and skips
+leased entries. Abandoned private partials count toward the limit and can be reclaimed.
+All lock waits and copying check the caller's absolute deadline; cleanup uses its allowance.
+No graph node mutation, executable deserialization or repository writes are introduced.
+
+## Status Report
+- STATUS: READY_FOR_REVIEW
+- TASK: 6.2
+- RED_PHASE_OUTPUT: `/tmp/group6-2-red.log`: seven tests failed with the temporary cache switch OFF. ON passed; switch removed; nine final cache tests passed.
+- TESTS_RUN: 13 cache/audit tests passed, full suite 1256 passed and 9 skipped (104.11s); ruff and mypy passed. Final receipt-validation repair was followed by the 13-test focused rerun.
+- EVIDENCE: concurrent subprocess writers, corrupted payload/manifest, interrupted atomic replace, abandoned partial files, bounded lock contention, leased eviction, unsafe paths and preservation of consumer exceptions.
+
+## Review Verdict
+- VERDICT: APPROVED
+- TASK: 6.2
+- MECHANICAL_RESULTS: canonical suite and final focused regression passed; static checks and diff whitespace passed; no new placeholders or secrets.
+- FINDINGS: review added metadata receipt validation and fixed contextmanager exception ownership; lock inodes are retained in 64 bounded stripes to avoid unlink/recreate races. Storage size means bytes of payloads/manifests, not filesystem allocation overhead.
+- SUMMARY: Complete publication or explicit miss is preserved through tested concurrency, corruption and interruption paths.
+
+## Verification Result
+- STATUS: VERIFIED
+- CLAIM_TYPE: TASK
+- CLAIM: Task 6.2 bounded local artifact publication and eviction.
+- EVIDENCE: nine cache contract tests and full suite above; production cache implementation, not a mocked storage path.
+- GAPS: Semantic keys and live replay integration follow in 6.3 and 6.4; no warm-latency claim.

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

@@ -167,7 +167,7 @@ hook latency is admitted.
   - Done when identical valid inputs can reuse a graph while any changed graph input or failed census rejects the lease; original cached artifacts survive query loading unchanged.
   - _Boundary: Cache and Engine Boundary, generic driver integration_
   - _Requirements: 4.2, 4.3, 4.4, 7.3_
-- [ ] 6.2 Publish and evict local artifacts safely
+- [x] 6.2 Publish and evict local artifacts safely
   - Add private temporary publication, per-key bounded locking, atomic completed manifests, cache size enforcement and unleased eviction.
   - Retain partial records only as diagnostics; never interpret a directory or failed write as a complete result. Do not rebuild graphs incrementally by mutating old node identities.
   - Done when concurrent writers, corrupt entries and interrupted publication yield either a validated complete artifact or an explicit miss without exceeding the remaining lock budget.

benchmarks/measurements/2026-09-08-module-audit.json

@@ -1,10 +1,10 @@
 {
   "counts": {
-    "load_bearing": 102,
+    "load_bearing": 103,
     "standalone_capability": 9
   },
-  "modules": 111,
-  "note": "Refreshed for task 5.1: explicit local replay integrates the existing ranker and comparison policy.",
+  "modules": 112,
+  "note": "Refreshed for task 6.2: bounded local artifact storage, integrated by task 6.4.",
   "orphaned": [],
   "rows": [
     {
@@ -128,6 +128,7 @@
       "classification": "load_bearing",
       "importers": [
         "cli",
+        "cpg.artifact",
         "cpg.backend",
         "model.config_value",
         "model.dominance",
@@ -136,6 +137,7 @@
         "model.pipeline",
         "model.scan",
         "model.taint",
+        "push.cache",
         "push.policy",
         "push.runner"
       ],
@@ -144,7 +146,10 @@
     },
     {
       "classification": "load_bearing",
-      "importers": [],
+      "importers": [
+        "cpg.backend",
+        "push.cache"
+      ],
       "module": "cpg.artifact",
       "reason": "pre-push-safety-net Req 4.3/7.3: backend graph identity and live lease contract; validation integration pending"
     },
@@ -152,6 +157,7 @@
       "classification": "load_bearing",
       "importers": [
         "cli",
+        "cpg.artifact",
         "model.config_value",
         "model.dominance",
         "model.judge",
@@ -159,6 +165,7 @@
         "model.pipeline",
         "model.scan",
         "model.taint",
+        "push.cache",
         "push.policy",
         "push.runner"
       ],
@@ -363,6 +370,7 @@
         "model.specs",
         "model.taint",
         "preprocess",
+        "push.cache",
         "push.contracts",
         "push.policy",
         "push.report",
@@ -409,6 +417,7 @@
         "model.partitions",
         "model.regions",
         "model.scan",
+        "push.cache",
         "push.contracts",
         "push.policy",
         "push.report",
@@ -687,6 +696,12 @@
       "module": "push",
       "reason": "pre-push-safety-net Req 5.5/7.3: experimental push contract package; no hook installed"
     },
+    {
+      "classification": "load_bearing",
+      "importers": [],
+      "module": "push.cache",
+      "reason": "pre-push-safety-net task 6.2: bounded artifact publication and leased eviction; runner integration in 6.4"
+    },
     {
       "classification": "load_bearing",
       "importers": [

src/openultrasast/model/audit.py

@@ -28,6 +28,7 @@ ROOTS = ("gate", "map_gate", "pair_gate", "cli", "__main__")
 
 # Modules that stand alone: their own CLI subcommand or external protocol, deliberately off the scan path.
 SPEC_OWNED = {
+    "push.cache": "pre-push-safety-net task 6.2: bounded artifact publication and leased eviction; runner integration in 6.4",
     "push.report": "pre-push-safety-net Req 5.1/5.2/5.5/7.3: compact output and complete artifacts; runner integration pending",
     "push.policy": "pre-push-safety-net Req 2.4/3.1/3.4/4.1: targeted same-semantics base comparison; replay integration pending",
     "push.snapshot": "pre-push-safety-net Req 1.1/1.3/1.4/7.1: local immutable ref resolution; runner integration pending",

src/openultrasast/push/cache.py

@@ -0,0 +1,217 @@
+"""Private local artifacts with bounded locks, complete publication and leased eviction."""
+
+from __future__ import annotations
+
+import fcntl
+import hashlib
+import json
+import os
+import re
+import stat
+import tempfile
+import time
+from collections.abc import Iterator
+from contextlib import ExitStack, contextmanager
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from openultrasast.cpg.artifact import digest_value, graph_bytes, read_bytes
+from openultrasast.cpg.backend import _DeadlineExpired, _remove_owned_tree
+from openultrasast.model.contracts import ExecutionBudget
+
+_KEY = re.compile(r"[0-9a-f]{64}")
+
+
+@dataclass(frozen=True)
+class CacheEntry:
+    path: Path
+    metadata: dict[str, Any]
+
+
+class ArtifactCache:
+    def __init__(self, root: Path, *, max_bytes: int) -> None:
+        if max_bytes <= 0:
+            raise ValueError("cache_size_must_be_positive")
+        if root.is_symlink():
+            raise ValueError("cache_root_symlink")
+        root.mkdir(parents=True, exist_ok=True, mode=0o700)
+        info = root.stat()
+        if info.st_uid != os.getuid() or info.st_mode & 0o077:
+            raise ValueError("cache_root_must_be_private_and_owned")
+        self.root = root.resolve()
+        self.max_bytes = max_bytes
+        self.last_reason = ""
+        self.hits = 0
+        self.misses = 0
+
+    @contextmanager
+    def _lock(self, name: str, budget: ExecutionBudget, *, shared: bool = False, wait: bool = True) -> Iterator[bool]:
+        fd = None
+        acquired = False
+        try:
+            fd = os.open(self.root / name, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600)
+            if not stat.S_ISREG(os.fstat(fd).st_mode):
+                raise ValueError("cache_lock_not_regular")
+            while time.monotonic() < budget.deadline_monotonic:
+                try:
+                    fcntl.flock(fd, (fcntl.LOCK_SH if shared else fcntl.LOCK_EX) | fcntl.LOCK_NB)
+                    acquired = True
+                    break
+                except BlockingIOError:
+                    if not wait:
+                        break
+                    time.sleep(min(0.01, max(0, budget.deadline_monotonic - time.monotonic())))
+            yield acquired
+        finally:
+            if fd is not None:
+                if acquired:
+                    fcntl.flock(fd, fcntl.LOCK_UN)
+                os.close(fd)
+
+    @staticmethod
+    def _bucket(key: str) -> str:
+        # Fixed lock stripes bound lock-file growth; never unlink a live lock inode.
+        return f".lock-{int(key[:2], 16) % 64:02d}"
+
+    def _entry(self, key: str, budget: ExecutionBudget) -> CacheEntry | None:
+        directory = self.root / key
+        if directory.is_symlink() or not directory.is_dir():
+            return None
+        manifest_path = directory / "manifest.json"
+        if manifest_path.stat().st_size > 1024 * 1024:
+            return None
+        manifest = json.loads(read_bytes(manifest_path, budget.deadline_monotonic))
+        if (
+            not isinstance(manifest, dict)
+            or manifest.get("version") != 1
+            or manifest.get("key") != key
+            or manifest.get("complete") is not True
+        ):
+            return None
+        receipt = manifest.pop("receipt", None)
+        if receipt != digest_value(manifest):
+            return None
+        if not isinstance(manifest.get("metadata"), dict):
+            return None
+        payload = directory / "payload"
+        if payload.stat().st_size != manifest.get("size") or payload.stat().st_size > self.max_bytes:
+            return None
+        if graph_bytes(payload, budget.deadline_monotonic) != manifest.get("sha256"):
+            return None
+        return CacheEntry(payload, manifest["metadata"])
+
+    @contextmanager
+    def lookup(self, key: str, budget: ExecutionBudget) -> Iterator[CacheEntry | None]:
+        entry = None
+        if not _KEY.fullmatch(key):
+            self.misses += 1
+            self.last_reason = "invalid_cache_key"
+            yield None
+            return
+        with ExitStack() as stack:
+            try:
+                locked = stack.enter_context(self._lock(self._bucket(key), budget, shared=True))
+                if locked:
+                    entry = self._entry(key, budget)
+            except (OSError, ValueError, TimeoutError):
+                entry = None
+            if entry is None:
+                self.misses += 1
+                self.last_reason = "cache_miss_or_incompatible"
+            else:
+                self.hits += 1
+            yield entry
+
+    def _remove(self, root: Path, budget: ExecutionBudget) -> None:
+        _remove_owned_tree(root, budget.deadline_monotonic)
+
+    def _make_room(self, incoming: int, key: str, budget: ExecutionBudget) -> bool:
+        entries = []
+        used = 0
+        for path in self.root.iterdir():
+            if time.monotonic() >= budget.deadline_monotonic:
+                return False
+            if path.is_symlink():
+                continue
+            if path.is_dir() and (_KEY.fullmatch(path.name) or path.name.startswith(".pending-")):
+                size = 0
+                for child in path.iterdir():
+                    if time.monotonic() >= budget.deadline_monotonic:
+                        return False
+                    size += child.lstat().st_size
+                used += size
+                entries.append((path.stat().st_mtime, path, size))
+        for _, path, size in sorted(entries):
+            if used + incoming <= self.max_bytes:
+                return True
+            if path.name == key or path.name.startswith(".pending-"):
+                self._remove(path, budget)
+                used -= size
+            else:
+                with self._lock(self._bucket(path.name), budget, wait=False) as locked:
+                    if locked:
+                        self._remove(path, budget)
+                        used -= size
+        return used + incoming <= self.max_bytes
+
+    def publish(self, key: str, payload: bytes | Path, metadata: dict[str, Any], budget: ExecutionBudget, *, complete: bool = True) -> bool:
+        if not complete or not _KEY.fullmatch(key):
+            self.last_reason = "partial_or_invalid_cache_publication"
+            return False
+        pending = None
+        try:
+            with self._lock(self._bucket(key), budget) as locked:
+                if not locked:
+                    return False
+                with self._lock(".publication-lock", budget) as publishing:
+                    if not publishing:
+                        return False
+                    size = len(payload) if isinstance(payload, bytes) else payload.stat().st_size
+                    if not size or size > self.max_bytes:
+                        return False
+                    sha = (
+                        hashlib.sha256(payload).hexdigest()
+                        if isinstance(payload, bytes)
+                        else graph_bytes(payload, budget.deadline_monotonic)
+                    )
+                    record = {"version": 1, "key": key, "complete": True, "metadata": metadata, "size": size, "sha256": sha}
+                    record["receipt"] = digest_value(record)
+                    manifest = json.dumps(record, sort_keys=True).encode()
+                    if len(manifest) > 1024 * 1024 or not self._make_room(size + len(manifest), key, budget):
+                        return False
+                    pending = Path(tempfile.mkdtemp(prefix=".pending-", dir=self.root))
+                    target = pending / "payload"
+                    if isinstance(payload, bytes):
+                        with target.open("xb") as stream:
+                            for offset in range(0, len(payload), 1024 * 1024):
+                                if time.monotonic() >= budget.deadline_monotonic:
+                                    raise TimeoutError("deadline_exhausted")
+                                stream.write(payload[offset : offset + 1024 * 1024])
+                    elif graph_bytes(payload, budget.deadline_monotonic, target) != sha:
+                        raise ValueError("publication_source_changed")
+                    target.chmod(0o600)
+                    with target.open("rb") as stream:
+                        os.fsync(stream.fileno())
+                    with (pending / "manifest.json").open("xb") as stream:
+                        stream.write(manifest)
+                        stream.flush()
+                        os.fsync(stream.fileno())
+                    (pending / "manifest.json").chmod(0o600)
+                    if time.monotonic() >= budget.deadline_monotonic:
+                        raise TimeoutError("deadline_exhausted")
+                    existing = self.root / key
+                    if existing.exists() or existing.is_symlink():
+                        self._remove(existing, budget)
+                    os.replace(pending, existing)
+                    pending = None
+                    return True
+        except (OSError, ValueError, TimeoutError, _DeadlineExpired) as error:
+            self.last_reason = type(error).__name__
+            return False
+        finally:
+            if pending is not None:
+                try:
+                    _remove_owned_tree(pending, budget.deadline_monotonic + budget.cancellation_allowance_seconds)
+                except (OSError, _DeadlineExpired):
+                    self.last_reason = "cache_partial_cleanup_incomplete"

tests/test_push_cache.py

@@ -0,0 +1,148 @@
+"""Publication, locking, corruption and eviction are observable cache contracts."""
+
+import time
+
+from openultrasast.cpg.artifact import digest_value
+from openultrasast.model.contracts import ExecutionBudget
+from openultrasast.push.cache import ArtifactCache
+
+
+def budget(seconds=3):
+    return ExecutionBudget(time.monotonic() + seconds, 0.2)
+
+
+def test_completed_publication_is_readable_and_private(tmp_path):
+    cache = ArtifactCache(tmp_path / "cache", max_bytes=10000)
+    key = digest_value("one")
+    assert cache.publish(key, b"real bytes", {"kind": "test"}, budget())
+    with cache.lookup(key, budget()) as entry:
+        assert entry is not None
+        assert entry.path.read_bytes() == b"real bytes"
+        assert entry.metadata == {"kind": "test"}
+        assert entry.path.stat().st_mode & 0o077 == 0
+    assert cache.root.stat().st_mode & 0o077 == 0
+
+
+def test_partial_and_corrupt_entries_are_misses(tmp_path):
+    cache = ArtifactCache(tmp_path / "cache", max_bytes=10000)
+    key = digest_value("one")
+    assert not cache.publish(key, b"partial", {}, budget(), complete=False)
+    with cache.lookup(key, budget()) as entry:
+        assert entry is None
+    assert cache.publish(key, b"good", {}, budget())
+    (cache.root / key / "payload").write_bytes(b"bad")
+    with cache.lookup(key, budget()) as entry:
+        assert entry is None
+    assert cache.publish(key, b"repaired", {}, budget())
+    (cache.root / key / "manifest.json").write_text("{}")
+    with cache.lookup(key, budget()) as entry:
+        assert entry is None
+
+
+def test_eviction_skips_leases_and_enforces_limit(tmp_path):
+    cache = ArtifactCache(tmp_path / "cache", max_bytes=1700)
+    one, two = digest_value("one"), digest_value("two")
+    assert cache.publish(one, b"a" * 1000, {}, budget())
+    with cache.lookup(one, budget()) as entry:
+        assert entry is not None
+        assert not cache.publish(two, b"b" * 1000, {}, budget(0.05))
+        assert entry.path.read_bytes() == b"a" * 1000
+    assert cache.publish(two, b"b" * 1000, {}, budget())
+    with cache.lookup(one, budget()) as entry:
+        assert entry is None
+    with cache.lookup(two, budget()) as entry:
+        assert entry is not None
+
+
+def test_lock_wait_cannot_exceed_transaction_budget(tmp_path):
+    cache = ArtifactCache(tmp_path / "cache", max_bytes=10000)
+    key = digest_value("one")
+    assert cache.publish(key, b"x", {}, budget())
+    with cache.lookup(key, budget()) as entry:
+        assert entry is not None
+        started = time.monotonic()
+        assert not cache.publish(key, b"y", {}, budget(0.05))
+        assert time.monotonic() - started < 0.25
+    with cache.lookup(key, budget()) as entry:
+        assert entry.path.read_bytes() == b"x"
+
+
+def test_interrupted_publication_never_creates_complete_entry(tmp_path, monkeypatch):
+    import os
+
+    cache = ArtifactCache(tmp_path / "cache", max_bytes=10000)
+    key = digest_value("one")
+
+    def fail(*args):
+        raise OSError("interrupted before atomic publication")
+
+    monkeypatch.setattr(os, "replace", fail)
+    assert not cache.publish(key, b"x", {}, budget())
+    with cache.lookup(key, budget()) as entry:
+        assert entry is None
+    assert not list(cache.root.glob(".pending-*"))
+
+
+def test_concurrent_writers_publish_only_complete_payloads(tmp_path):
+    import multiprocessing
+
+    cache = ArtifactCache(tmp_path / "cache", max_bytes=10000)
+    key = digest_value("one")
+
+    def write(value):
+        assert cache.publish(key, value * 2000, {"writer": value.decode()}, budget())
+
+    processes = [multiprocessing.get_context("fork").Process(target=write, args=(v,)) for v in (b"a", b"b")]
+    for process in processes:
+        process.start()
+    for process in processes:
+        process.join(5)
+        assert process.exitcode == 0
+    with cache.lookup(key, budget()) as entry:
+        assert entry.path.read_bytes() == entry.metadata["writer"].encode() * 2000
+
+
+def test_symlink_and_invalid_key_are_rejected(tmp_path):
+    import pytest
+
+    target = tmp_path / "target"
+    target.mkdir()
+    link = tmp_path / "link"
+    link.symlink_to(target)
+    with pytest.raises(ValueError):
+        ArtifactCache(link, max_bytes=10000)
+    cache = ArtifactCache(tmp_path / "cache", max_bytes=10000)
+    with cache.lookup("../escape", budget()) as entry:
+        assert entry is None
+    assert not cache.publish("../escape", b"x", {}, budget())
+    key = digest_value("link")
+    (cache.root / key).symlink_to(target)
+    with cache.lookup(key, budget()) as entry:
+        assert entry is None
+
+
+def test_lookup_failure_is_miss_and_consumer_exception_is_preserved(tmp_path):
+    import pytest
+
+    cache = ArtifactCache(tmp_path / "cache", max_bytes=10000)
+    key = digest_value("one")
+    assert cache.publish(key, b"bytes", {}, budget())
+    with pytest.raises(OSError, match="consumer"), cache.lookup(key, budget()) as entry:
+        assert entry is not None
+        raise OSError("consumer")
+    lock = cache.root / cache._bucket(key)
+    lock.unlink()
+    lock.symlink_to(tmp_path / "absent")
+    with cache.lookup(key, budget()) as entry:
+        assert entry is None
+
+
+def test_abandoned_partial_files_count_toward_size_and_are_evicted(tmp_path):
+    cache = ArtifactCache(tmp_path / "cache", max_bytes=1700)
+    abandoned = cache.root / ".pending-dead-writer"
+    abandoned.mkdir(mode=0o700)
+    (abandoned / "payload").write_bytes(b"a" * 1000)
+    key = digest_value("one")
+    assert cache.publish(key, b"b" * 1000, {}, budget())
+    assert not abandoned.exists()
+    assert sum(p.stat().st_size for p in cache.root.rglob("*") if p.is_file()) <= cache.max_bytes