feat(pre-push-safety-net): measure frozen PMPro and Node replay feasibility

benchmarks/push/README.md

@@ -89,3 +89,51 @@ results, per-capability metrics, coverage and cold/warm latency. Missing benign
 have unknown interruption rate. VAmPI's readable regression may be measured while
 independent admission prerequisites remain missing. Unsupported/unreviewed populations
 stay visible. All results remain experimental pending the later joint admission gate.
+
+## First full replay experiment (task 5.2)
+
+`measure.py` runs the six prepared PMPro/NodeGoat security, fixed and benign cases
+through the production `ousast pre-push` CLI. It does not supply a hand-selected
+source slice or regions. The complete 11-case population remains in the scorer;
+Ghost is not scanned, and the separate VAmPI/libpng populations are retained as
+unexecuted regression/envelope evidence rather than being silently dropped.
+
+Build the local image first, then run with a writable output parent and the existing
+repository-object cache mounted read-only:
+
+```sh
+docker run --rm --network none --memory 3g --entrypoint python \
+  -v "$PWD:/context:ro" \
+  -v "$HOME/.cache/openultrasast/repos:/cache:ro" \
+  -v /path/to/output-parent:/results \
+  openultrasast:dev /context/benchmarks/push/measure.py \
+  --manifest /context/benchmarks/push/inputs.json --cache /cache --out /results/new-run
+```
+
+The output directory must not already exist. Input validation opens and verifies
+pinned source/license blobs before any measurement. A private bare Git repository
+reads the existing objects and creates reproducible authored commits using only the
+specified edits. Every other tracked blob remains in the tree; the cache's files,
+index and refs are unchanged. No project code, package installation or network is used.
+
+`profile.json` is written before the first replay. PMPro runs first, followed by an
+explicit `freeze-after-pmpro.json` checkpoint and Node with identical installed core,
+facts, queries and policy. Both analysis and cancellation allowances stay 30 and 2
+seconds. Whole CLI elapsed time, separate available stage costs, exact input pins,
+source byte counts, selected/deferred identities, raw artifacts, process exits and
+terminal output remain available. `rank` denotes an observed selected-question position;
+an unlocated/unselected target retains null. A completed empty query cannot earn a
+witnessed negative or fixed-side silence. No transitive path is fabricated from rank.
+
+`scorecard.json` uses the existing frozen-population scorer. `decision.json` explicitly
+returns NO-GO because this initial cold diagnostic does not qualify independent
+capabilities or prove joint quality and warm latency. Exit **1** with a completed
+`decision.json` means measured NO-GO; a traceback without that completion record is a
+failed instrument. Zero CLI exit means advisory allow, never complete analysis.
+Missing artifacts and unanswered questions remain errors/unresolved, not clean results.
+
+Cold here means no persisted graph/query/comparison reuse; filesystem caches are not
+flushed. Six one-shot runs are diagnostic samples, not representative warm p95 evidence.
+Normal capabilities remain disabled, so zero alerts do not establish precision or useful
+recall. A separately named longer lab run may obtain witnesses, but cannot replace the
+30-second measurements or satisfy the hook's latency gate.

benchmarks/push/measure.py

@@ -0,0 +1,313 @@
+"""First frozen-core PMPro/Node diagnostic experiment, using local Git objects only."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import platform
+import subprocess
+import tempfile
+import time
+from dataclasses import replace
+from pathlib import Path
+
+from openultrasast.model.contracts import QuestionIdentity
+from openultrasast.model.scan import ScanBudget
+from openultrasast.push.runner import _provenance
+from openultrasast.push_inputs import validate_manifest
+from openultrasast.push_scoring import freeze_profile, score
+from openultrasast.repos import checkout_path, load_repo_recipe
+
+
+def git(root: Path, *args: str, data: bytes | None = None) -> bytes:
+    env = {key: value for key, value in os.environ.items() if not key.startswith("GIT_")}
+    env.update(
+        GIT_CONFIG_GLOBAL=os.devnull,
+        GIT_CONFIG_NOSYSTEM="1",
+        GIT_NO_REPLACE_OBJECTS="1",
+        GIT_NO_LAZY_FETCH="1",
+        GIT_ALLOW_PROTOCOL="",
+        GIT_GRAFT_FILE=os.devnull,
+        GIT_AUTHOR_NAME="Authored evaluation",
+        GIT_AUTHOR_EMAIL="[email protected]",
+        GIT_COMMITTER_NAME="Authored evaluation",
+        GIT_COMMITTER_EMAIL="[email protected]",
+        GIT_AUTHOR_DATE="2026-09-12T00:00:00+00:00",
+        GIT_COMMITTER_DATE="2026-09-12T00:00:00+00:00",
+    )
+    return subprocess.run(
+        ["git", "-c", "core.hooksPath=" + os.devnull, "-C", str(root), *args],
+        input=data,
+        capture_output=True,
+        check=True,
+        timeout=30,
+        env=env,
+    ).stdout
+
+
+def pin_snapshot(bare: Path, commit: str, snapshot: dict) -> str:
+    """Apply declared edits to a private Git index; retain every other tracked blob."""
+    git(bare, "read-tree", commit)
+    edited = False
+    for entry in snapshot["files"]:
+        content = git(bare, "cat-file", "blob", commit + ":" + entry["path"])
+        if not content:
+            raise ValueError("empty input bytes")
+        if edit := entry.get("edit"):
+            before, after = edit["before"].encode(), edit["after"].encode()
+            if not edit["provenance"].startswith("authored") or not before or content.count(before) != 1:
+                raise ValueError("authored edit must match exactly once")
+            if hashlib.sha256(content).hexdigest() != entry["original_sha256"]:
+                raise ValueError("original hash mismatch")
+            content = content.replace(before, after, 1)
+            edited = True
+        if len(content) != entry["size"] or hashlib.sha256(content).hexdigest() != entry["sha256"]:
+            raise ValueError("resulting input identity mismatch")
+        if entry.get("edit"):
+            blob = git(bare, "hash-object", "-w", "--stdin", data=content).decode().strip()
+            mode = git(bare, "ls-tree", commit, "--", entry["path"]).split(b" ", 1)[0].decode()
+            git(bare, "update-index", "--add", "--cacheinfo", mode, blob, entry["path"])
+    if not edited:
+        return commit
+    tree = git(bare, "write-tree").decode().strip()
+    return (
+        git(bare, "commit-tree", tree, "-p", commit, data=("Authored evaluation snapshot: " + snapshot["id"] + "\n").encode())
+        .decode()
+        .strip()
+    )
+
+
+def execution_record(case_id: str, artifact: dict, *, elapsed: float, target: dict | None = None) -> dict:
+    """Associate existing answers; empty rows cannot manufacture a witnessed negative."""
+    selected, deferred, queries = [], [], []
+    rank = None
+    modes = set()
+    for record in artifact["scans"]:
+        if record["side"] != "head":
+            continue
+        scan = record["scan"]
+        if scope := scan.get("scope"):
+            modes.add(scope["ranking_mode"])
+            if target is not None:
+                rank = next(
+                    (
+                        index
+                        for index, q in enumerate(scope["selected"], 1)
+                        if all(q["identity"][key] == target[key] for key in ("path", "function", "family"))
+                    ),
+                    None,
+                )
+            selected.extend(QuestionIdentity.from_payload(q["identity"]).question_id for q in scope["selected"])
+            deferred.extend(QuestionIdentity.from_payload(q["identity"]).question_id for q in scope["deferred"])
+        for outcome in scan["question_outcomes"]:
+            identity = QuestionIdentity.from_payload(outcome["identity"])
+            query = {
+                "id": identity.question_id,
+                "status": "answered" if outcome["status"] == "completed" else "unanswered",
+                "outcome": "unknown",
+            }
+            # Keep the exact question/answer association. A raw finding reached from
+            # another function is not a proved transitive path for this scorer.
+            findings = [
+                f
+                for f in scan["findings"]
+                if f["family"] == identity.family
+                and f["site"].startswith(identity.path + ":")
+                and f["site"].rsplit(":", 1)[-1] == identity.function
+            ]
+            if outcome["status"] == "completed" and findings:
+                target_sites = {f"{target['path']}:{line}:{target['function']}" for line in target["lines"]} if target else set()
+                finding = next((f for f in findings if f["site"] in target_sites), findings[0])
+                query.update(outcome="positive", finding={**finding, "site": finding["site"].rsplit(":", 1)[0]})
+            queries.append(query)
+    if modes and modes != {"evidence"}:
+        raise ValueError("runtime ranker differs from frozen evidence mode")
+    deltas = [d["candidate"]["delta"]["novelty"] for d in artifact["admission"]["dispositions"]]
+    novelty = next((d for d in ("new", "worsened", "unknown", "unchanged") if d in deltas), "unknown")
+    reasons = artifact["admission"]["coverage_reasons"]
+    coverage = (
+        "timeout"
+        if any("deadline" in reason for reason in reasons)
+        else ("complete" if artifact["result"]["coverage_status"] == "complete_within_scope" else "incomplete")
+    )
+    if artifact["admission"]["defects"]:
+        raise ValueError("unqualified experiment unexpectedly emitted an admitted alert")
+    return {
+        "case_id": case_id,
+        "ranking_mode": "evidence",
+        "rank": rank,
+        "rank_kind": "selected_question_position",
+        "selected": selected,
+        "deferred": deferred,
+        "queries": queries,
+        "alerts": [],
+        "delta": novelty,
+        "coverage": coverage,
+        "cache_state": "cold",
+        "timings": {
+            "total": elapsed,
+            "build": sum(r["scan"].get("build_seconds", 0) for r in artifact["scans"]),
+            "query": sum(r["scan"].get("query_seconds", 0) for r in artifact["scans"]),
+        },
+    }
+
+
+def write(path: Path, data: dict) -> None:
+    path.write_text(json.dumps(data, indent=2, allow_nan=False) + "\n")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--manifest", type=Path, required=True)
+    parser.add_argument("--cache", type=Path, required=True)
+    parser.add_argument("--out", type=Path, required=True)
+    args = parser.parse_args()
+    args.out.mkdir(parents=True, exist_ok=False)
+    manifest = json.loads(args.manifest.read_bytes())
+    validation = validate_manifest(args.manifest, cache_root=args.cache, fetch=False)
+    write(args.out / "input-validation.json", validation)
+    settings = ScanBudget(max_model_calls=0, max_regions=500, order_by_evidence=True)
+    initial = _provenance(args.manifest.parent, settings)
+    keys = ("ranking_mode", "engine", "facts", "queries", "policy", "core")
+    config = {key: initial[key] for key in keys}
+    config.update(
+        version="first-replay-v2-2026-09-13",
+        scope="full tracked tree; declared first-party graph/target exclusions; evidence ranker",
+        hardware=platform.platform(),
+        budget_seconds=30,
+        targets=json.loads((args.manifest.parent / "diagnostic-config-v1.json").read_text())["targets"],
+    )
+    profile = freeze_profile(args.manifest.read_bytes(), config)
+    write(args.out / "profile.json", profile)  # Freeze on disk BEFORE any measurement.
+    run = {"profile_sha256": profile["profile_sha256"], "context": dict(config), "records": []}
+    evidence = {"cases": [], "core_changes": [], "adapter_changes": [], "freeze_after_pmpro": None}
+    ready = {c["id"]: c["status"] == "ready" for c in validation["cases"]}
+    snapshots = {s["id"]: s for s in manifest["snapshots"]}
+    # Fixed declared order; reserved, regression and unsupported populations stay
+    # in the joint scorer without scanning or tuning on the holdout.
+    cases = [c for prefix in ("pmpro-", "nodegoat-") for c in manifest["cases"] if c["id"].startswith(prefix)]
+    with tempfile.TemporaryDirectory(prefix="ousast-experiment-") as temporary:
+        bare = Path(temporary) / "objects.git"
+        bare.mkdir()
+        git(bare, "init", "--bare")
+        sources = {}
+        for case in cases:
+            for sid in (case["base"], case["tip"]):
+                if sid in sources or not ready[case["id"]]:
+                    continue
+                snapshot = snapshots[sid]
+                recipe = load_repo_recipe(args.manifest.parent / snapshot["recipe"])
+                recipe = replace(recipe, commit=snapshot.get("commit", recipe.commit))
+                source = checkout_path(recipe, args.cache)
+                objects = git(source, "rev-parse", "--path-format=absolute", "--git-path", "objects").decode().strip()
+                sources[sid] = (recipe.commit, objects)
+        (bare / "objects/info/alternates").write_text("\n".join(sorted({value[1] for value in sources.values()})) + "\n")
+        pins = {sid: pin_snapshot(bare, commit, snapshots[sid]) for sid, (commit, _) in sources.items()}
+        write(
+            args.out / "replay-pins.json",
+            {"snapshots": pins, "provenance": "Full upstream trees; declared authored edits in a private bare Git repository."},
+        )
+        for case in cases:
+            if not ready[case["id"]]:
+                evidence["cases"].append({"case_id": case["id"], "status": "missing_input_prerequisite"})
+                continue
+            current = _provenance(args.manifest.parent, settings)
+            if any(current[key] != initial[key] for key in keys):
+                raise ValueError("core/facts/query/policy changed after freezing")
+            if case["id"].startswith("nodegoat-") and evidence["freeze_after_pmpro"] is None:
+                evidence["freeze_after_pmpro"] = {key: current[key] for key in keys}
+                write(args.out / "freeze-after-pmpro.json", evidence["freeze_after_pmpro"])
+            artifact_path = args.out / (case["id"] + ".json")
+            command = [
+                "ousast",
+                "pre-push",
+                str(bare),
+                "--base",
+                pins[case["base"]],
+                "--head",
+                pins[case["tip"]],
+                "--artifact",
+                str(artifact_path),
+                "--deadline",
+                "30",
+                "--cancellation-allowance",
+                "2",
+                "--max-regions",
+                "500",
+            ]
+            start = time.monotonic()
+            process = subprocess.run(command, capture_output=True, text=True, timeout=62)
+            elapsed = time.monotonic() - start
+            (args.out / (case["id"] + ".stdout")).write_text(process.stdout)
+            (args.out / (case["id"] + ".stderr")).write_text(process.stderr)
+            entry = {
+                "case_id": case["id"],
+                "command": command,
+                "process_exit_code": process.returncode,
+                "elapsed_seconds": elapsed,
+                "artifact": artifact_path.name if artifact_path.exists() else None,
+                "terminal": process.stdout,
+                "core_unchanged": True,
+                "input_pins": {"base": pins[case["base"]], "head": pins[case["tip"]]},
+            }
+            if artifact_path.exists():
+                artifact = json.loads(artifact_path.read_text())
+                # Unmeasured provenance caused by preparation timeout is explicit;
+                # never replace it with a fictitious runtime measurement.
+                entry["runtime_provenance_available"] = all(artifact["provenance"].get(key) == initial[key] for key in keys)
+                record = execution_record(case["id"], artifact, elapsed=elapsed, target=config["targets"][case["id"]])
+                entry["coverage_reasons"] = artifact["admission"]["coverage_reasons"]
+                entry["stage_timings"] = artifact["timings"]
+                entry["snapshot_bytes_read"] = sum(s["bytes_read"] for s in artifact["snapshots"])
+                entry["classification"] = "scope/deadline" if record["coverage"] == "timeout" else "recorded context/capability limits"
+            else:
+                record = {
+                    "case_id": case["id"],
+                    "ranking_mode": "evidence",
+                    "rank": None,
+                    "selected": [],
+                    "deferred": [],
+                    "queries": [],
+                    "alerts": [],
+                    "delta": "unknown",
+                    "coverage": "error",
+                    "cache_state": "cold",
+                    "timings": {"total": elapsed, "build": 0, "query": 0},
+                }
+                entry["classification"] = "instrument/artifact unavailable; zero stage costs mean unmeasured"
+            if process.returncode != 0:
+                entry["classification"] = "instrument/CLI failure"
+                record["coverage"] = "error"
+            run["records"].append(record)
+            evidence["cases"].append(entry)
+            write(args.out / "run.json", run)
+            write(args.out / "execution-evidence.json", evidence)
+            print(json.dumps({"case": case["id"], "coverage": record["coverage"], "elapsed_seconds": elapsed}), flush=True)
+    scored = score(profile, validation, run)
+    write(args.out / "scorecard.json", scored)
+    decision = {
+        "decision": "NO-GO",
+        "capabilities_enabled": False,
+        "scorecard": "scorecard.json",
+        "reason": "First cold diagnostic experiment; independent eligibility and joint quality/warm-latency gates remain unmet.",
+        "recorded_cases": len(run["records"]),
+        "declared_population": len(manifest["cases"]),
+        "core_unchanged": True,
+        "adapter_changes": [],
+        "limits": [
+            "No warm-cache measurements",
+            "Untouched Ghost population not scanned",
+            "PHP/Python admission populations missing",
+            "libpng retained as unsupported envelope; no C/C++ detection claim",
+            "Zero normal alerts from an empty registry cannot establish precision or useful recall",
+        ],
+    }
+    write(args.out / "decision.json", decision)
+    print(json.dumps(decision), flush=True)
+    return 1  # Measurement completed; the feasibility/admission decision is explicitly NO-GO.
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())