Merge unseen-freeze: freeze p1 (bisect guard, freeze writer, protocol)

Co-Authored-By: Claude Opus 4.8 <[email protected]>

.kiro/specs/unseen-repo-evaluation/tasks.md

@@ -130,14 +130,14 @@ Execution rules:
   - _Requirements: 2.1, 2.2_
 
 - [ ] 3. Freeze p1 (design 1.2, 6; increment 3)
-- [ ] 3.1 Guard bisection tool
+- [x] 3.1 Guard bisection tool
   - `benchmarks/unseen/bisect_guard.py`: rewrites the draft with a candidate subset, runs
     `.venv/bin/pytest -q -p no:cacheprovider tests/test_independent_population.py -k referenced` with stdout and
     stderr discarded, reads only the exit code, and bisects until green. Reports the dropped count only.
   - `tests/test_unseen_bisect.py` with an injected fake guard (a hidden set): finds and drops exactly the hidden
     members; never logs a candidate; an exit code other than 0 or 1 aborts (instrument failure).
   - _Requirements: 1.1_
-- [ ] 3.2 Freeze writer
+- [x] 3.2 Freeze writer
   - `benchmarks/unseen/freeze.py`: canonical `pool-p1.toml` (and the private counterpart) from the bisected draft;
     `freeze-p1.json` with used-set digest and per-source counts, guard result, dropped count, label-check counts
     and the freeze digest (sha256 of the canonical manifest). Refuses to write over a frozen pool; a correction
@@ -145,7 +145,7 @@ Execution rules:
   - Tests: digest stable under key order; overwrite refused; the guard's digest test (1.3) passes on a fixture
     pool.
   - _Requirements: 1.2_
-- [ ] 3.3 Protocol p1
+- [x] 3.3 Protocol p1
   - `benchmarks/unseen/protocol-p1.md`: the scoring rules of design section 4 (default flags, decision 4.0's
     long-deadline arm, catch and location rules, false alarm, coverage states), the statistics of section 3 and
     7, and the adoption gate of 7.3, stated before any replay.

benchmarks/unseen/bisect_guard.py

@@ -0,0 +1,170 @@
+"""Stage a cache draft, bisect with the black-box guard, and retain a digest receipt.
+
+Coordinator: python -m benchmarks.unseen.bisect_guard --draft CACHE/draft-p1.toml
+The cache is never changed. Successful subsets remain at benchmarks/unseen/draft-p1.toml
+and private/draft-p1.toml; failures remove only this invocation's staging files.
+Stdout reports only the dropped count. No reservation is opened by this module.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import subprocess
+import tomllib
+from pathlib import Path
+
+from .draft import manifest, private_digest
+from .extract import InstrumentFailure
+from .journal import atomic, digest
+from .source import PERMISSIVE
+
+
+def read_manifest(path: Path) -> dict:
+    try:
+        raw = path.read_bytes()
+        if not raw:
+            raise ValueError
+        data = tomllib.loads(raw.decode("utf-8"))
+        if set(data) - {"status", "pool", "seed", "repository"}:
+            raise ValueError
+        if data["status"] != "draft" or data["pool"] != "p1" or type(data["seed"]) is not int:
+            raise ValueError
+        if not isinstance(data.get("repository", []), list):
+            raise ValueError
+        return data
+    except (OSError, ValueError, KeyError, TypeError):
+        raise InstrumentFailure("invalid_draft") from None
+
+
+def read_pair(public_path: Path, private_path: Path) -> tuple[dict, dict]:
+    """Validate the licensing split and every private pointer before staging names."""
+    public, private = read_manifest(public_path), read_manifest(private_path)
+    try:
+        if public["seed"] != private["seed"]:
+            raise ValueError
+        hidden = {row["id"]: row for row in private.get("repository", [])}
+        rows = public.get("repository", [])
+        if len(hidden) != len(private.get("repository", [])) or len({r["id"] for r in rows}) != len(rows):
+            raise ValueError
+        if set(hidden) != {r["id"] for r in rows if r.get("private")}:
+            raise ValueError
+        for row in rows:
+            if type(row["slice"]) is not int or row["slice"] not in (1, 2, 3):
+                raise ValueError
+            if row.get("private"):
+                full = hidden[row["id"]]
+                expected = {"id": full["id"], "slice": full["slice"], "private": True, "sha256": private_digest(full)}
+                if row != expected or full["license_class"] != "private":
+                    raise ValueError
+            elif row["license_class"] != "permissive" or row["license_spdx"] not in PERMISSIVE:
+                raise ValueError
+        return public, private
+    except (ValueError, KeyError, TypeError):
+        raise InstrumentFailure("invalid_draft_pair") from None
+
+
+def pytest_guard(root: Path) -> int:
+    """Read only the exit code; never capture, inspect or forward either stream."""
+    try:
+        return subprocess.run(
+            [".venv/bin/pytest", "-q", "-p", "no:cacheprovider", "tests/test_independent_population.py", "-k", "referenced"],
+            cwd=root,
+            stdin=subprocess.DEVNULL,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            check=False,
+        ).returncode
+    except OSError:
+        raise InstrumentFailure("guard_launch") from None
+
+
+def run(root: Path, draft: Path, *, private_draft: Path | None = None, guard_runner=None) -> dict:
+    """Inject guard_runner(root) -> exit code for offline tests; default is pytest_guard.
+
+    Run from an otherwise clean reservation directory. Existing staging/receipt or a
+    frozen p1 is refused, so a crash cannot silently turn a partial subset into input.
+    The receipt binds both retained manifests; freeze refuses edits after the guard.
+    """
+    private_draft = private_draft or draft.with_name("private-draft-p1.toml")
+    public, private = read_pair(draft, private_draft)
+    rows = public.get("repository", [])
+    if not rows:
+        raise InstrumentFailure("empty_draft")
+    directory = root / "benchmarks/unseen"
+    public_path = directory / "draft-p1.toml"
+    private_path = directory / "private/draft-p1.toml"
+    receipt_path = directory / "bisect-p1.json"
+    targets = (private_path, public_path, receipt_path)
+    if any(p.exists() for p in (*targets, directory / "pool-p1.toml", directory / "freeze-p1.json")):
+        raise FileExistsError("staging_exists")
+    private_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+    runner = guard_runner or pytest_guard
+    created = []
+
+    def check(subset):
+        ids = {r["id"] for r in subset}
+        selected = [r for r in private.get("repository", []) if r["id"] in ids]
+        atomic(private_path, manifest(selected, private["seed"]).encode("utf-8"))
+        atomic(public_path, manifest(subset, public["seed"]).encode("utf-8"))
+        try:
+            code = runner(root)
+        except Exception:
+            raise InstrumentFailure("guard_launch") from None
+        if type(code) is not int or code not in (0, 1):
+            raise InstrumentFailure("guard_exit")
+        return code
+
+    def retain(subset):
+        if not subset or check(subset) == 0:
+            return subset
+        if len(subset) == 1:
+            return []
+        middle = len(subset) // 2
+        return retain(subset[:middle]) + retain(subset[middle:])
+
+    try:
+        # Reserve destinations exclusively before any rewrites; never truncate an
+        # existing coordinator's files. Empty manifests establish the control.
+        for path in targets:
+            with path.open("xb") as handle:
+                created.append(path)
+                os.fchmod(handle.fileno(), 0o600)
+        if check([]) != 0:
+            raise InstrumentFailure("guard_baseline")
+        kept = retain(rows)
+        if check(kept) != 0:
+            raise InstrumentFailure("guard_final")
+        result = {
+            "guard_result": 0,
+            "dropped_count": len(rows) - len(kept),
+            "input_count": len(rows),
+            "draft_digest": digest(read_manifest(public_path)),
+            "private_draft_digest": digest(read_manifest(private_path)),
+        }
+        atomic(receipt_path, (json.dumps(result, sort_keys=True) + "\n").encode())
+        return result
+    except BaseException:
+        for path in reversed(created):
+            path.unlink(missing_ok=True)
+        raise
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--root", type=Path, default=Path.cwd())
+    parser.add_argument("--draft", type=Path, required=True)
+    parser.add_argument("--private-draft", type=Path)
+    args = parser.parse_args(argv)
+    try:
+        result = run(args.root, args.draft, private_draft=args.private_draft)
+    except Exception:
+        print(json.dumps({"status": "instrument_failure"}))
+        return 1
+    print(json.dumps({"dropped_count": result["dropped_count"]}))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

benchmarks/unseen/freeze.py

@@ -0,0 +1,150 @@
+"""Freeze the staged, bisected p1 without running a guard, git, network or model.
+
+Coordinator: python -m benchmarks.unseen.freeze --used-report USED-SET-REPORT.json
+The report is eligibility.UsedSet.report() output. Label-check counts are derived
+from retained vulnerability changes' label_check fields (missing means unchecked).
+A matching retry is a no-op; any changed frozen artifact requires a new pool id.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import tomllib
+from pathlib import Path
+
+from .bisect_guard import read_pair
+from .draft import manifest
+from .eligibility import SOURCES
+from .journal import digest, sync_directory
+
+
+def count(value) -> int:
+    if type(value) is not int or value < 0:
+        raise ValueError("invalid_count")
+    return value
+
+
+def sha256(value) -> str:
+    if not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value):
+        raise ValueError("invalid_digest")
+    return value
+
+
+def metadata(used_report: dict) -> dict:
+    """Only known source labels and numeric counts can enter the public record."""
+    try:
+        used_digest = sha256(used_report["sha256"])
+        sources = used_report["sources"]
+        if not sources or set(sources) - {*SOURCES, "memory_local", "memory_s3"}:
+            raise ValueError
+        return {"used_set_digest": used_digest, "source_counts": {k: count(v["count"]) for k, v in sources.items()}}
+    except (ValueError, KeyError, TypeError, AttributeError):
+        raise ValueError("invalid_used_report") from None
+
+
+def verify_receipt(receipt: dict, public: dict, private: dict) -> None:
+    try:
+        if type(receipt["guard_result"]) is not int or receipt["guard_result"] != 0:
+            raise ValueError
+        if sha256(receipt["draft_digest"]) != digest(public) or sha256(receipt["private_draft_digest"]) != digest(private):
+            raise ValueError
+        if count(receipt["input_count"]) != len(public.get("repository", [])) + count(receipt["dropped_count"]):
+            raise ValueError
+    except (ValueError, KeyError, TypeError):
+        raise ValueError("invalid_guard_receipt") from None
+
+
+def frozen_text(data: dict) -> bytes:
+    # Reuse the draft serializer, changing only the status. Hash its parsed value,
+    # not the TOML bytes: table/key order and whitespace are not pool identity.
+    text = manifest(data.get("repository", []), data["seed"])
+    return text.replace('status = "draft"', 'status = "frozen"', 1).encode("utf-8")
+
+
+def matches(path: Path, payload: bytes) -> bool:
+    try:
+        if path.suffix == ".toml":
+            return digest(tomllib.loads(path.read_text(encoding="utf-8"))) == digest(tomllib.loads(payload.decode("utf-8")))
+        return digest(json.loads(path.read_bytes())) == digest(json.loads(payload))
+    except (OSError, ValueError):
+        return False
+
+
+def run(root: Path, *, used_report: dict) -> dict:
+    directory = root / "benchmarks/unseen"
+    record = metadata(used_report)
+    public, private = read_pair(directory / "draft-p1.toml", directory / "private/draft-p1.toml")
+    try:
+        receipt = json.loads((directory / "bisect-p1.json").read_bytes())
+    except (OSError, ValueError):
+        raise ValueError("invalid_guard_receipt") from None
+    verify_receipt(receipt, public, private)
+    if not public.get("repository"):
+        raise ValueError("empty_pool")
+    labels = dict.fromkeys(("confirmed", "refactor", "unclear", "unchecked"), 0)
+    hidden = {r["id"]: r for r in private.get("repository", [])}
+    try:
+        for pointer in public["repository"]:
+            row = hidden[pointer["id"]] if pointer.get("private") else pointer
+            for change in row["changes"]:
+                if change["kind"] == "vulnerability":
+                    labels[change.get("label_check", "unchecked")] += 1
+    except (KeyError, TypeError):
+        raise ValueError("invalid_label_check") from None
+    try:
+        protocol = (directory / "protocol-p1.md").read_bytes()
+        if not protocol:
+            raise ValueError
+    except (OSError, ValueError):
+        raise ValueError("invalid_protocol") from None
+    public_bytes, private_bytes = frozen_text(public), frozen_text(private)
+    record.update(
+        guard_result=0,
+        dropped_count=receipt["dropped_count"],
+        label_check_counts=labels,
+        freeze_digest=digest(tomllib.loads(public_bytes.decode("utf-8"))),
+        protocol_digest=hashlib.sha256(protocol).hexdigest(),
+    )
+    outputs = (
+        (directory / "private/pool-p1.toml", private_bytes),
+        (directory / "pool-p1.toml", public_bytes),
+        (directory / "freeze-p1.json", (json.dumps(record, sort_keys=True, indent=2) + "\n").encode()),
+    )
+    # Check EVERY existing artifact before writing anything. The record is the
+    # completion marker and is written last; matching interrupted writes can resume.
+    if any(path.exists() and not matches(path, payload) for path, payload in outputs):
+        raise FileExistsError("frozen_pool_differs_use_new_pool")
+    for path, payload in outputs:
+        try:
+            with path.open("xb") as handle:
+                os.fchmod(handle.fileno(), 0o600)
+                handle.write(payload)
+                handle.flush()
+                os.fsync(handle.fileno())
+            sync_directory(path.parent)
+        except FileExistsError:
+            if not matches(path, payload):
+                raise FileExistsError("frozen_pool_differs_use_new_pool") from None
+    return record
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--root", type=Path, default=Path.cwd())
+    parser.add_argument("--used-report", type=Path, required=True)
+    args = parser.parse_args(argv)
+    try:
+        report = run(args.root, used_report=json.loads(args.used_report.read_bytes()))
+    except Exception:
+        print(json.dumps({"status": "freeze_refused"}))
+        return 1
+    print(json.dumps(report, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

benchmarks/unseen/protocol-p1.md

@@ -0,0 +1,180 @@
+# Pool p1: frozen scoring protocol
+
+This protocol fixes the scoring, uncertainty estimates and adoption gate before any
+p1 replay. Its authority is the approved
+[unseen-repo-evaluation design](../../.kiro/specs/unseen-repo-evaluation/design.md),
+sections 3, 4 and 7 (including decisions 4.0, 4.0a and 7.3). The freeze record binds
+these exact UTF-8 file bytes by `protocol_digest`. Corrections require a new pool;
+neither p1 nor this protocol is revised in response to results.
+
+## Population, slices and use
+
+Design section 3 fixes three seeded, ecosystem- and `post_cutoff`-stratified slices
+of 100 repositories: one vulnerability change and 12 ordinary changes per
+repository, 100 + 1,200 = 1,300 changes per slice, 300 repositories and 3,900 changes
+overall. Guard-rejected candidates are dropped before freezing; actual retained
+counts are reported, with no replacement or silent claim that reduced slices meet
+the planned precision. Frozen slices are never resized.
+
+Slice 1 supplies the baseline and G1 arms table. Slices 2 and 3 qualify the first
+two adoption decisions. Once spent, p2 is drawn by the same code and excludes p1.
+Every use is appended to `usage-p1.jsonl` with slice, decision or experiment id,
+purpose (`baseline`, `exploratory`, `informed`, `qualifies`), date, result digest and
+freeze digest. A slice that informed a change cannot qualify that change (3.1).
+
+The label check reads every third `introducing` change before freezing, marking
+`confirmed`, `refactor` or `unclear`; report the confirmed share and catch on the
+confirmed subset. Report catch separately for `introducing` (the headline) and
+`last_touch` (a lower bound). The former inherits SZZ last-writer/refactor noise;
+the latter may already contain the flaw in base, making novelty-based silence
+correct even though scoring counts a miss (design 2.1).
+
+## Replay arms
+
+The default hook replay uses `--deadline 30 --engine inline --max-regions 500
+--mode advisory` (section 4). Retain the artifact and full text output, including
+every `Skipped:` line.
+
+Decision 4.0 registers a second slice-1 arm with `--deadline 300` and every other
+flag unchanged. Commit its experiment manifest with
+`unseen: {pool: p1, slice: 1}` before its first replay. Report both arms to
+distinguish deadline misses from engine misses. This arm is a measurement;
+adopting a longer default deadline is a separate decision under section 7.2.
+
+Decision 4.0a calibrates cluster overhead against native Docker runc. Use a
+deterministic paired sample of 20 changes by default, or all of a smaller proof
+manifest, at a 600-second calibration deadline. Record per-change stage times and
+median cluster/native slowdown with bootstrap 95% intervals separately for time
+through quick, engine build and engine queries. Incomplete timings or degraded
+scans invalidate calibration. Cluster arm deadlines are the 30/300-second
+deadlines multiplied by the median paired slowdown through quick, rounded to
+integer seconds; native deadlines stay at 30/300 seconds. Report applied
+deadlines and factors.
+
+A deterministic native sample accompanies the cluster slice. Compare whether
+each push flags anything and report a Wilson 95% agreement interval. If its lower
+bound is below 95%, or any sample pair is missing or failed, flag the run and
+exclude cluster results from the baseline. Identify disagreements by opaque
+change IDs only. Calibration records contain counts, times and opaque digests.
+Docker checks its data-root free space against the unpacked image size upper
+bound plus 1 GiB before explicitly pulling; retain sanitized stderr tails and
+exit codes on failure (4.0a).
+
+## Catch and false alarm
+
+A vulnerability change is caught only if a finding matches its labelled family
+and head location (4.1). Engine findings carry the family; quick-rule CWEs map
+through the project's family CWE table. Match locations in the corresponding
+file using the following rules:
+
+| Family | Required head location |
+| --- | --- |
+| `injection`, `path`, `deserialization`, `output_encoding`, `untrusted_destination` | Sink line in the vulnerable function, or within 10 lines of the head-side changed hunks |
+| `access_control` | The vulnerable function itself (the handler lacking the check), or within 10 lines of the head-side changed hunks |
+| `config_secrets` | The changed file and head-side changed lines ±10, at file scope (`<global>`) |
+
+All three user-visible streams count: admitted defects, advisory engine findings
+and quick-rule matches. Record the stream responsible for each catch. A matching
+family at the wrong location does not count.
+
+An ordinary change is a false alarm if **any** of those three streams contains at
+least one finding, of any family, anywhere (4.2). Report the admitted-defects-only
+alerts rate beside it. An ordinary change may contain an unknown vulnerability;
+adjudication does not change this definition. Adjudicate all flagged ordinary
+changes if there are 30 or fewer, otherwise every third, against the standard
+of protocol v2, and report the adjudicated share separately (design 2.2).
+
+## Coverage and degradation
+
+Assign exactly one state to every change (4.3):
+
+| State | Definition |
+| --- | --- |
+| `instrument_failure` | No record, failed fetch, head commit not verified, or zero bytes of changed files read; also the `suspect_fast` failure path |
+| `unanalysable` | No changed file uses a language the hook covers (`language_not_covered`) |
+| `quick_only` | Quick read the changed files but no engine question completed on them |
+| `engine_covered` | At least one engine question completed on a changed file |
+
+Prove input access: record checkout files and bytes, bytes of each changed file,
+JVM wall time and hook exit code. A replay faster than JVM startup is
+`suspect_fast`, never a clean zero. Failed fetches, missing commits and hook
+crashes yield failure records, never empty successes. Re-run an instrument
+failure once, then report it; it never counts as quiet or missed. Refuse to score
+a slice with more than 5% instrument failures after re-runs (design Error
+Handling). Preserve skipped/degraded coverage rather than interpreting silence
+as analysis. Report catch and false alarm overall and by coverage state, with
+failure counts explicit and failures excluded from outcome denominators.
+
+## Intervals and planned precision
+
+Use Wilson 95% intervals with z = 1.96 (section 3). For observed proportion p and
+denominator n, the centre is `(p + z²/(2n)) / (1 + z²/n)` and the half width is
+`z * sqrt(p*(1-p)/n + z²/(4n²)) / (1 + z²/n)`. The precision bar uses the larger
+distance from p to either bound, not just the interval's half width.
+
+The design's exact minimum sample sizes are:
+
+| Rate | Assumed p | Maximum distance | Required n |
+| --- | --- | --- | --- |
+| False alarm | 0.05 | 3 percentage points | 315 |
+| False alarm | 0.10 | 3 percentage points | 483 |
+| False alarm | 0.20 | 3 percentage points | 756 |
+| Catch | 0.10 | 10 percentage points | 62 |
+| Catch | 0.25 | 10 percentage points | 88 |
+| Catch | 0.40 | 10 percentage points | 97 |
+
+At n = 100 and catch p = 0.4 the Wilson interval is [0.309, 0.498], within
+10 points. One vulnerability change per repository keeps those observations
+independent. The 12 ordinary changes within a repository are correlated. With
+m = 12 and assumed ICC = 0.05 the design effect `1 + (m - 1)*ICC` is 1.55;
+1,200 ordinary changes give effective n ≈774 ≥756. At p = 0.2 their plain Wilson
+interval is about ±2.3 points.
+
+Report a repository-cluster bootstrap 95% interval beside the ordinary-change
+Wilson interval: resample whole repositories, retaining their within-repository
+changes. Measure and record ICC on slice 1. If ICC exceeds 0.05, increase the next
+pool's size, never the frozen slice's. Per-family intervals are reported but not
+sized; splitting 100 changes over seven families gives roughly ±20–25 points.
+The design leaves per-family BLOCK sizing to a later decision.
+
+Report counts, denominators, rates and intervals overall and per family, method,
+post-cutoff group, stream, coverage state and confirmed subset (sections 3 and 7).
+Use actual observed counts, not planned sample sizes, for intervals.
+
+## Fixed budgets and adoption gate
+
+For score-bearing arms, report catch at false-alarm budgets of 1%, 5% and 10%
+(7.1). Choose the lowest score threshold whose ordinary-change false-alarm rate
+stays within the budget; report catch with its Wilson interval and realised
+false-alarm rate. A threshold selected on the same slice is `exploratory`. For
+qualification, fix it beforehand on an earlier slice or held-out folds and
+apply it unchanged. A binary arm has one operating point; report which budgets
+it meets. Do not substitute AUC for these operating points.
+
+Registered experiments identify `unseen: {pool, slice}`. Write the unseen
+estimates beside held-out fold estimates, including each arm's catch at each
+budget and paired difference B − A with an exact McNemar test on the paired
+vulnerability changes (planned n = 100; 7.2). Record use in the ledger.
+
+Adopt candidate B over production A only when both the existing fold rule and
+the untouched-slice gate pass (7.3):
+
+1. At the 5% budget with B's threshold fixed beforehand, B's catch is not
+   significantly lower than A's: exact two-sided McNemar on paired vulnerability
+   changes, alpha 0.05.
+2. For a binary arm, B's false-alarm rate is not significantly higher than A's:
+   exact two-sided McNemar on paired ordinary changes, alpha 0.05. Report the
+   repository-cluster bootstrap interval for the false-alarm difference beside
+   the test because those changes are clustered.
+
+Write `unseen.gate: pass | regression | not_run`. No slice, a slice already
+`informed` for this change, or more than 5% instrument failures yields `not_run`;
+it blocks adoption just as `regression` does. Decision 4.0a also excludes flagged
+cluster/native agreement runs from baseline use. Failure to detect a significant
+regression is the specified gate, not evidence that the arms are equivalent.
+
+Raw per-change artifacts remain outside the repository. Committed measurement
+records contain counts, rates, intervals and digests, never repository names or
+URLs. Public p1 names stay only in the permissive pool manifest; private entries
+remain in the gitignored private manifest and are bound through opaque pointers
+(design 1.4 and 6).

tests/test_unseen_bisect.py

@@ -0,0 +1,192 @@
+"""Offline guard controls: the hidden reservation exists only inside the fake runner."""
+
+import importlib
+import json
+import subprocess
+import sys
+import tomllib
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+d = importlib.import_module("benchmarks.unseen.draft")
+b = importlib.import_module("benchmarks.unseen.bisect_guard")
+
+
+def draft_pair(tmp_path, count=9):
+    root, cache = tmp_path / "tree", tmp_path / "cache"
+    root.mkdir()
+    cache.mkdir()
+    rows = []
+    for i in range(count):
+        changes = [{"kind": "vulnerability", "method": "introducing", "base": "a" * 40, "head": "b" * 40}]
+        changes[0]["label_check"] = ("confirmed", "refactor", "unclear")[i % 3]
+        changes += [{"kind": "ordinary", "base": f"{j:040x}", "head": f"{j + 20:040x}"} for j in range(12)]
+        rows.append(
+            {
+                "repository": f"synthetic/candidate-{i}",
+                "url": "https://" + f"github.com/synthetic/candidate-{i}",
+                "license_class": "private" if i % 2 else "permissive",
+                "license_spdx": "GPL-3.0" if i % 2 else "MIT",
+                "slice": i % 3 + 1,
+                "changes": changes,
+            }
+        )
+        rows[-1]["id"] = "r-" + d.private_digest(rows[-1])[:24]
+    public, private = [], []
+    for row in rows:
+        if row["license_class"] == "private":
+            private.append(row)
+            public.append({"id": row["id"], "slice": row["slice"], "private": True, "sha256": d.private_digest(row)})
+        else:
+            public.append(row)
+    draft = cache / "draft-p1.toml"
+    draft.write_text(d.manifest(public, 19))
+    (cache / "private-draft-p1.toml").write_text(d.manifest(private, 19))
+    return root, draft, rows
+
+
+def staged(root):
+    directory = root / "benchmarks/unseen"
+    public = tomllib.loads((directory / "draft-p1.toml").read_text())
+    private = tomllib.loads((directory / "private/draft-p1.toml").read_text())
+    hidden = {row["id"]: row for row in private.get("repository", [])}
+    rows = [hidden[row["id"]] if row.get("private") else row for row in public.get("repository", [])]
+    assert set(hidden) == {row["id"] for row in public.get("repository", []) if row.get("private")}
+    return rows
+
+
+@pytest.mark.parametrize("members", [(), (0,), (1, 4, 7), tuple(range(9))])
+def test_bisects_hidden_members_only(tmp_path, capsys, caplog, members):
+    root, draft, rows = draft_pair(tmp_path)
+    original = draft.read_bytes()
+    reserved = {rows[i]["repository"] for i in members}
+    calls = []
+
+    def guard(root):
+        names = {row["repository"] for row in staged(root)}
+        calls.append(names)
+        return int(bool(names & reserved))
+
+    result = b.run(root, draft, guard_runner=guard)
+    assert result["dropped_count"] == len(members)
+    assert result["guard_result"] == 0
+    assert {r["repository"] for r in staged(root)} == {r["repository"] for r in rows} - reserved
+    assert calls[0] == set() and not (calls[-1] & reserved)
+    assert draft.read_bytes() == original
+    assert capsys.readouterr() == ("", "") and not caplog.records
+    receipt = (root / "benchmarks/unseen/bisect-p1.json").read_text()
+    for row in rows:
+        assert row["repository"] not in receipt and row["url"] not in receipt
+    assert json.loads(receipt) == result
+
+
+@pytest.mark.parametrize("code", [-9, 2, 3, 4, 5, 127])
+def test_unknown_exit_aborts_and_removes_staging(tmp_path, code):
+    root, draft, _ = draft_pair(tmp_path)
+    calls = 0
+
+    def guard(root):
+        nonlocal calls
+        calls += 1
+        return 0 if calls == 1 else code
+
+    with pytest.raises(b.InstrumentFailure, match="guard_exit"):
+        b.run(root, draft, guard_runner=guard)
+    assert not list((root / "benchmarks/unseen").rglob("*.toml"))
+    assert not (root / "benchmarks/unseen/bisect-p1.json").exists()
+
+
+def test_red_empty_baseline_is_not_candidate_rejection(tmp_path):
+    root, draft, _ = draft_pair(tmp_path)
+    with pytest.raises(b.InstrumentFailure, match="guard_baseline"):
+        b.run(root, draft, guard_runner=lambda root: 1)
+
+
+def test_final_combination_is_checked(tmp_path):
+    root, draft, _ = draft_pair(tmp_path)
+    codes = iter([0, 0, 1])
+    with pytest.raises(b.InstrumentFailure, match="guard_final"):
+        b.run(root, draft, guard_runner=lambda root: next(codes))
+
+
+def test_guard_subprocess_is_exact_and_discards_both_streams(tmp_path, monkeypatch):
+    calls = []
+
+    def fake(argv, **kwargs):
+        calls.append((argv, kwargs))
+        return subprocess.CompletedProcess(argv, 1)
+
+    monkeypatch.setattr(b.subprocess, "run", fake)
+    assert b.pytest_guard(tmp_path) == 1
+    argv, kwargs = calls[0]
+    assert argv == [".venv/bin/pytest", "-q", "-p", "no:cacheprovider", "tests/test_independent_population.py", "-k", "referenced"]
+    assert kwargs["cwd"] == tmp_path
+    assert kwargs["stdout"] == kwargs["stderr"] == subprocess.DEVNULL
+
+
+def test_existing_staging_is_never_overwritten(tmp_path):
+    root, draft, _ = draft_pair(tmp_path)
+    path = root / "benchmarks/unseen/draft-p1.toml"
+    path.parent.mkdir(parents=True)
+    path.write_bytes(b"preserve")
+    with pytest.raises(FileExistsError):
+        b.run(root, draft, guard_runner=lambda root: pytest.fail("guard should not run"))
+    assert path.read_bytes() == b"preserve"
+
+
+def test_cli_outputs_dropped_count_only(tmp_path, monkeypatch, capsys):
+    root, draft, _ = draft_pair(tmp_path)
+    monkeypatch.setattr(b, "pytest_guard", lambda root: 0)
+    assert b.main(["--root", str(root), "--draft", str(draft)]) == 0
+    assert json.loads(capsys.readouterr().out) == {"dropped_count": 0}
+
+
+def test_cli_failure_does_not_echo_exception(tmp_path, monkeypatch, capsys):
+    root, draft, rows = draft_pair(tmp_path)
+
+    def guard(root):
+        raise OSError(rows[0]["url"])
+
+    monkeypatch.setattr(b, "pytest_guard", guard)
+    assert b.main(["--root", str(root), "--draft", str(draft)]) == 1
+    assert capsys.readouterr() == ('{"status": "instrument_failure"}\n', "")
+
+
+@pytest.mark.parametrize("problem", ["empty_file", "empty_pool", "missing_private", "bad_pointer", "private_in_public"])
+def test_invalid_draft_never_runs_guard(tmp_path, problem):
+    root, draft, rows = draft_pair(tmp_path)
+    private = draft.with_name("private-draft-p1.toml")
+    if problem == "empty_file":
+        draft.write_bytes(b"")
+    elif problem == "empty_pool":
+        draft.write_text(d.manifest([], 19))
+        private.write_text(d.manifest([], 19))
+    elif problem == "missing_private":
+        private.unlink()
+    elif problem == "bad_pointer":
+        data = tomllib.loads(draft.read_text())
+        data["repository"][1]["sha256"] = "0" * 64
+        draft.write_text(d.manifest(data["repository"], 19))
+    else:
+        draft.write_text(d.manifest(rows, 19))
+    with pytest.raises(b.InstrumentFailure):
+        b.run(root, draft, guard_runner=lambda root: pytest.fail("guard should not run"))
+    assert not (root / "benchmarks/unseen/draft-p1.toml").exists()
+
+
+def test_interrupt_removes_partial_subset(tmp_path):
+    root, draft, _ = draft_pair(tmp_path)
+    calls = 0
+
+    def guard(root):
+        nonlocal calls
+        calls += 1
+        if calls > 2:
+            raise KeyboardInterrupt
+        return 0 if calls == 1 else 1
+
+    with pytest.raises(KeyboardInterrupt):
+        b.run(root, draft, guard_runner=guard)
+    assert not list((root / "benchmarks/unseen").rglob("*.toml"))

tests/test_unseen_freeze.py

@@ -0,0 +1,197 @@
+"""Freeze fixture pools without reading reservations or running subprocesses."""
+
+import hashlib
+import importlib
+import json
+import tomllib
+from pathlib import Path
+
+import pytest
+from test_unseen_bisect import b, d, draft_pair
+
+f = importlib.import_module("benchmarks.unseen.freeze")
+
+
+def prepared(tmp_path):
+    root, draft, rows = draft_pair(tmp_path)
+    b.run(root, draft, guard_runner=lambda root: 0)
+    protocol = root / "benchmarks/unseen/protocol-p1.md"
+    protocol.write_bytes(Path("benchmarks/unseen/protocol-p1.md").read_bytes())
+    return root, rows
+
+
+def used():
+    return {"sha256": "a" * 64, "sources": {"pairs": {"count": 7}, "memory_s3": {"count": 3}}}
+
+
+def test_freeze_split_receipt_and_existing_guard_digest(tmp_path, capsys, caplog, monkeypatch):
+    root, rows = prepared(tmp_path)
+    record = f.run(root, used_report=used())
+    directory = root / "benchmarks/unseen"
+    pool = directory / "pool-p1.toml"
+    public = tomllib.loads(pool.read_text())
+    private = tomllib.loads((directory / "private/pool-p1.toml").read_text())
+    assert public["status"] == private["status"] == "frozen"
+    assert record["used_set_digest"] == "a" * 64
+    assert record["source_counts"] == {"pairs": 7, "memory_s3": 3}
+    assert record["label_check_counts"] == {"confirmed": 3, "refactor": 3, "unclear": 3, "unchecked": 0}
+    assert record["guard_result"] == 0 and record["dropped_count"] == 0
+    assert record["protocol_digest"] == hashlib.sha256((directory / "protocol-p1.md").read_bytes()).hexdigest()
+    assert record["freeze_digest"] == hashlib.sha256(d.canonical(public)).hexdigest()
+    assert json.loads((directory / "freeze-p1.json").read_text()) == record
+    for row in rows:
+        assert row["repository"] not in json.dumps(record) and row["url"] not in json.dumps(record)
+        if row["license_class"] == "private":
+            assert row["url"] not in pool.read_text() and row["repository"] not in pool.read_text()
+            pointer = next(p for p in public["repository"] if p["id"] == row["id"])
+            assert pointer == {"id": row["id"], "slice": row["slice"], "private": True, "sha256": d.private_digest(row)}
+            assert row in private["repository"]
+        else:
+            assert row in public["repository"]
+    assert (directory / "private/pool-p1.toml").stat().st_mode & 0o777 == 0o600
+    guard = importlib.import_module("test_independent_population")
+    monkeypatch.setattr(guard, "ROOT", root)
+    guard.test_frozen_pool_manifest_digest()
+    assert capsys.readouterr() == ("", "") and not caplog.records
+
+
+def test_digest_and_retry_ignore_toml_key_order(tmp_path):
+    root, _ = prepared(tmp_path)
+    original = f.run(root, used_report=used())
+    pool = root / "benchmarks/unseen/pool-p1.toml"
+    text = pool.read_text()
+    lines = text.splitlines()
+    pool.write_text("\n".join([lines[2], lines[1], lines[0], *lines[3:]]) + "\n")
+    before = pool.read_bytes()
+    assert f.run(root, used_report=used()) == original
+    assert pool.read_bytes() == before
+
+
+@pytest.mark.parametrize("target", ["pool-p1.toml", "private/pool-p1.toml", "freeze-p1.json"])
+def test_numeric_type_change_is_a_different_frozen_artifact(tmp_path, target):
+    root, _ = prepared(tmp_path)
+    f.run(root, used_report=used())
+    path = root / "benchmarks/unseen" / target
+    text = path.read_text()
+    if path.suffix == ".toml":
+        modified = text.replace("seed = 19", "seed = 19.0")
+    else:
+        modified = text.replace('"guard_result": 0', '"guard_result": false')
+    assert text != modified
+    path.write_text(modified)
+    with pytest.raises(FileExistsError, match="new_pool"):
+        f.run(root, used_report=used())
+    assert path.read_text() == modified
+
+
+@pytest.mark.parametrize("target", ["pool-p1.toml", "private/pool-p1.toml", "freeze-p1.json", "protocol-p1.md"])
+def test_refuses_different_frozen_files(tmp_path, target):
+    root, _ = prepared(tmp_path)
+    f.run(root, used_report=used())
+    directory = root / "benchmarks/unseen"
+    path = directory / target
+    if path.suffix == ".toml":
+        path.write_text(path.read_text().replace('status = "frozen"', 'status = "modified"'))
+    elif path.suffix == ".json":
+        record = json.loads(path.read_text())
+        record["dropped_count"] += 1
+        path.write_text(json.dumps(record))
+    else:
+        path.write_text("changed protocol")
+    before = {p: p.read_bytes() for p in directory.rglob("*") if p.is_file()}
+    with pytest.raises(FileExistsError, match="new_pool"):
+        f.run(root, used_report=used())
+    assert all(p.read_bytes() == value for p, value in before.items())
+
+
+@pytest.mark.parametrize("target", ["draft-p1.toml", "private/draft-p1.toml", "bisect-p1.json"])
+def test_refuses_stale_or_red_guard_receipt(tmp_path, target):
+    root, _ = prepared(tmp_path)
+    path = root / "benchmarks/unseen" / target
+    if target.endswith(".json"):
+        receipt = json.loads(path.read_text())
+        receipt["guard_result"] = 1
+        path.write_text(json.dumps(receipt))
+    else:
+        before = path.read_text()
+        after = before.replace('"head" = "', '"head" = "0', 1)
+        assert before != after
+        path.write_text(after)
+    with pytest.raises((ValueError, b.InstrumentFailure)):
+        f.run(root, used_report=used())
+    assert not (root / "benchmarks/unseen/pool-p1.toml").exists()
+
+
+@pytest.mark.parametrize("mutation", ["digest", "source_name", "count"])
+def test_rejects_identity_bearing_metadata(tmp_path, mutation):
+    root, rows = prepared(tmp_path)
+    report = used()
+    if mutation == "digest":
+        report["sha256"] = rows[0]["url"]
+    elif mutation == "source_name":
+        report["sources"][rows[0]["repository"]] = {"count": 1}
+    else:
+        report["sources"]["pairs"]["count"] = rows[0]["url"]
+    with pytest.raises(ValueError):
+        f.run(root, used_report=report)
+    assert not (root / "benchmarks/unseen/freeze-p1.json").exists()
+
+
+def test_missing_protocol_aborts_before_publication(tmp_path):
+    root, _ = prepared(tmp_path)
+    (root / "benchmarks/unseen/protocol-p1.md").unlink()
+    with pytest.raises(ValueError, match="protocol"):
+        f.run(root, used_report=used())
+    assert not (root / "benchmarks/unseen/pool-p1.toml").exists()
+
+
+def test_drops_flow_into_freeze_counts_and_private_manifest(tmp_path):
+    root, draft, rows = draft_pair(tmp_path)
+    hidden = {rows[i]["repository"] for i in (1, 2)}
+    from test_unseen_bisect import staged
+
+    b.run(root, draft, guard_runner=lambda root: int(any(r["repository"] in hidden for r in staged(root))))
+    directory = root / "benchmarks/unseen"
+    (directory / "protocol-p1.md").write_bytes(Path("benchmarks/unseen/protocol-p1.md").read_bytes())
+    record = f.run(root, used_report=used())
+    assert record["dropped_count"] == 2
+    assert record["label_check_counts"] == {"confirmed": 3, "refactor": 2, "unclear": 2, "unchecked": 0}
+    assert len(tomllib.loads((directory / "pool-p1.toml").read_text())["repository"]) == 7
+    assert len(tomllib.loads((directory / "private/pool-p1.toml").read_text())["repository"]) == 3
+
+
+def test_changed_private_labels_invalidate_guard_receipt(tmp_path):
+    root, _ = prepared(tmp_path)
+    path = root / "benchmarks/unseen/private/draft-p1.toml"
+    before = path.read_text()
+    after = before.replace('"label_check" = "refactor"', '"label_check" = "confirmed"', 1)
+    assert before != after
+    path.write_text(after)
+    with pytest.raises(ValueError, match="guard_receipt"):
+        f.run(root, used_report=used())
+
+
+def test_utf8_digest_matches_guard_with_non_ascii_metadata(tmp_path, monkeypatch):
+    root, draft, _ = draft_pair(tmp_path)
+    data = tomllib.loads(draft.read_text())
+    data["repository"][0]["license_path"] = "licence-ä.txt"
+    draft.write_text(d.manifest(data["repository"], 19))
+    b.run(root, draft, guard_runner=lambda root: 0)
+    (root / "benchmarks/unseen/protocol-p1.md").write_bytes(Path("benchmarks/unseen/protocol-p1.md").read_bytes())
+    f.run(root, used_report=used())
+    guard = importlib.import_module("test_independent_population")
+    monkeypatch.setattr(guard, "ROOT", root)
+    guard.test_frozen_pool_manifest_digest()
+
+
+def test_freeze_cli_redacts_failure_and_reports_only_record(tmp_path, capsys):
+    root, rows = prepared(tmp_path)
+    report = tmp_path / "used.json"
+    report.write_text(json.dumps(used()))
+    assert f.main(["--root", str(root), "--used-report", str(report)]) == 0
+    output = capsys.readouterr()
+    assert json.loads(output.out)["guard_result"] == 0 and output.err == ""
+    assert all(r["repository"] not in output.out for r in rows)
+    report.write_text(rows[0]["url"])
+    assert f.main(["--root", str(root), "--used-report", str(report)]) == 1
+    assert capsys.readouterr() == ('{"status": "freeze_refused"}\n', "")