From 6d27c9bfcf04c26a58feb5c784dc8d1967d505ce Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Mon, 5 Oct 2026 08:14:57 +0200 Subject: [PATCH] unseen-eval: calibrated replay deadline + native agreement check, docker-lane diagnostics, proof 4.4 record; engine runner cleans up fetched sources gVisor replays exhausted the hook's 30 s deadline in the snapshot step on a 1,725-file repository; cluster replays now use 30 s x a measured slowdown factor, verified on a native sample. The engine runner deletes each pin's fetched source (1.8 GB had accumulated), caps the cache at 500 MB and refuses below 1.5 GB free. Proof 4.4: two engine pins via the generic dispatcher match the VM baseline exactly. Implemented with Codex. Co-Authored-By: Claude Opus 5.5 --- .kiro/specs/unseen-repo-evaluation/design.md | 16 ++ benchmarks/ax/batch.py | 111 ++++++++- benchmarks/learn/engine_trace.py | 41 ++- .../record.json | 110 ++++++++ benchmarks/unseen/README.md | 81 ++++++ benchmarks/unseen/calibrate.py | 128 ++++++++++ benchmarks/unseen/proof.py | 2 +- benchmarks/unseen/replay.py | 82 ++++-- benchmarks/unseen/replay_entry.py | 9 +- benchmarks/unseen/score.py | 55 ++++ tests/test_engine_trace.py | 101 ++++++++ tests/test_unseen_calibrate.py | 235 ++++++++++++++++++ tests/test_unseen_proof.py | 30 +++ 13 files changed, 976 insertions(+), 25 deletions(-) create mode 100644 benchmarks/measurements/2026-10-04-unseen-04-dispatcher/record.json create mode 100644 benchmarks/unseen/calibrate.py create mode 100644 tests/test_unseen_calibrate.py diff --git a/.kiro/specs/unseen-repo-evaluation/design.md b/.kiro/specs/unseen-repo-evaluation/design.md index 4b7b492..76db1b3 100644 --- a/.kiro/specs/unseen-repo-evaluation/design.md +++ b/.kiro/specs/unseen-repo-evaluation/design.md @@ -219,6 +219,22 @@ what the user sees (30 s) and what the engine could catch given time (300 s), so told apart from a miss by the engine. It is a measurement arm, not a proposed default: adopting a longer hook deadline would be a separate decision under 7.2. +**Decision 4.0a: calibrated cluster deadline and native check** (maintainer, 2026-10-05). +The gVisor proof exhausted 30 seconds in snapshot preparation on a 1,725-file repository. +Cluster replays therefore scale the 30/300-second arms by the median paired cluster/native +time through the quick tier, rounded to integer seconds. Native runc replays retain 30/300 seconds. +A deterministic sample (default 20, or all of a smaller proof manifest) runs both lanes at +600 seconds for calibration. Report per-change stage times and median slowdown with bootstrap +95% intervals for time through quick, engine build and engine queries separately. Incomplete +timings or degraded scans invalidate calibration. Records contain counts, times and opaque digests. + +A deterministic native sample accompanies the cluster slice. Compare whether each push flags +anything, with a Wilson 95% interval. If its lower bound is below 95%, or any sample pair is +missing/failed, flag the run and exclude cluster results from the baseline. Disagreements name +change IDs only. The scorer reports applied deadlines and factors. Docker pulls explicitly after +checking its data-root free space against an unpacked image size upper bound plus 1 GiB; failures +retain sanitized stderr tails and exit codes. See the unseen README for coordinator commands. + **Decision 4.1: catch, per family.** A finding counts if its family is the change's family. Engine streams carry a family; a quick-rule match maps its CWE through the family CWE table. Its head location must also be in the vulnerable function or within 10 lines of the change's head-side hunks (protocol v1's window). diff --git a/benchmarks/ax/batch.py b/benchmarks/ax/batch.py index 039b584..b5feec1 100644 --- a/benchmarks/ax/batch.py +++ b/benchmarks/ax/batch.py @@ -387,7 +387,15 @@ def dispatch(items, workload, out, executors, *, lanes=("ax", "ax", "docker"), o for artifact in attempt_output.iterdir(): if artifact.is_file(): shutil.copyfile(artifact, output / artifact.name) - attempts.append({"lane": lane, "status": record.get("status"), "attempt": attempt, "transport": record.get("ax_attempts", [])}) + attempts.append( + { + "lane": lane, + "status": record.get("status"), + "attempt": attempt, + "transport": record.get("ax_attempts", []), + "docker_errors": record.get("docker_errors", []), + } + ) if record.get("status") not in {"sandbox_failure", "instrument_failure"}: break return {**record, "attempts": (previous or {}).get("attempts", []) + attempts, "input_digest": item.input_digest, "done": True} @@ -402,9 +410,76 @@ def dispatch(items, workload, out, executors, *, lanes=("ax", "ax", "docker"), o return schedule(pending, lanes, out, execute, complete, fallbacks=fallbacks, overflow=overflow) +def docker_diagnostic(text): + """Bound diagnostics and remove bearer URLs, environment secrets and assignments.""" + text = str(text or "") + for key, value in os.environ.items(): + if len(value) >= 4 and any(word in key.upper() for word in ("TOKEN", "SECRET", "PASSWORD", "CREDENTIAL", "KEY")): + text = text.replace(value, "[redacted]") + text = re.sub(r"https?://[^\s\"']+", "[redacted-url]", text) + text = re.sub( + r"(?i)(authorization|bearer|password|token|secret|signature|credential|[\w-]*key)\s*[:= ]\s*[^\s,;]+", r"\1=[redacted]", text + ) + return "\n".join(text.splitlines()[-10:])[-4000:] + + class DockerLane(AXLane): """VM lane uses the same presigned-object protocol and pinned replay image.""" + def __init__(self, *args, runner=None, **kwargs): + super().__init__(*args, **kwargs) + self.runner = runner or subprocess.run + self._prepared = False + + def __call__(self, item, output, attempt): + self._diagnostics = [] + try: + if not self._prepared: + self._prepare_image() + self._prepared = True + except (ValueError, OSError, subprocess.SubprocessError) as exc: + if not self._diagnostics: + self._diagnostics.append({"step": "preflight", "exit_code": None, "stderr": docker_diagnostic(str(exc))}) + record = {"status": "instrument_failure", "docker_errors": self._diagnostics} + write_json(output / "docker-error.json", record) + return record + try: + record = super().__call__(item, output, attempt) + finally: + if self._diagnostics: + write_json(output / "docker-error.json", {"docker_errors": self._diagnostics}) + if self._diagnostics: + record["docker_errors"] = self._diagnostics + return record + + def _run_docker(self, command, timeout): + try: + done = self.runner(command, capture_output=True, text=True, timeout=timeout) + except (OSError, subprocess.SubprocessError) as exc: + self._diagnostics.append( + {"step": command[1], "exit_code": None, "stderr": docker_diagnostic(getattr(exc, "stderr", "") or type(exc).__name__)} + ) + raise + if done.returncode: + self._diagnostics.append({"step": command[1], "exit_code": done.returncode, "stderr": docker_diagnostic(done.stderr)}) + raise ValueError(f"docker {command[1]} failed (exit {done.returncode})") + return done + + def _prepare_image(self): + # Docker's data root, not the coordinator checkout, is where layers consume space. + root = self._run_docker(["docker", "info", "--format", "{{.DockerRootDir}}"], 30).stdout.strip() + size = getattr(self.args, "docker_image_bytes", None) + if not isinstance(size, int) or size <= 0: + raise ValueError("set --docker-image-bytes to the unpacked image size upper bound before pulling") + free = self.runner(["df", "-B1", "--output=avail", root], capture_output=True, text=True, timeout=30) + if free.returncode: + raise ValueError("cannot check free space on Docker data root") + available = int(free.stdout.splitlines()[-1].strip()) + required = size + 1024**3 + if available < required: + raise ValueError(f"refusing Docker pull: free bytes {available} below image + 1 GiB ({required})") + self._run_docker(["docker", "pull", self.workload.image], 900) + def _execute_attempt(self, item, output, attempt, attempts): # Reuse all object/validation/cleanup logic, replacing only the task runner. import copy @@ -417,18 +492,44 @@ class DockerLane(AXLane): def _docker_command(self, *args, manifest=None, timeout=30): if args[0] == "apply": self._docker_name = manifest["metadata"]["name"] - command = ["docker", "run", "-d", "--name", self._docker_name, "--entrypoint", manifest["spec"]["command"][0]] + self._docker_created = True # Even a failed run can leave a created container. + command = [ + "docker", + "run", + "-d", + "--runtime", + "runc", + "--pull", + "never", + "--name", + self._docker_name, + "--entrypoint", + manifest["spec"]["command"][0], + ] for env in manifest["spec"]["env"]: command += ["-e", env["name"] + "=" + env["value"]] command += [manifest["spec"]["image"], *manifest["spec"]["command"][1:], "--heap-profile", "vm"] elif args[0] == "get": command = ["docker", "inspect", "--format", "{{.State.Running}} {{.State.ExitCode}}", self._docker_name] else: + if not getattr(self, "_docker_created", False): + return subprocess.CompletedProcess([], 0, "", "") command = ["docker", "rm", "-f", self._docker_name or args[2]] - done = subprocess.run(command, capture_output=True, text=True, timeout=timeout) - if done.returncode: - raise ValueError("docker command failed") + try: + done = self._run_docker(command, timeout) + except ValueError: + if args[0] == "delete" and "No such container" in self._diagnostics[-1]["stderr"]: + self._diagnostics.pop() + return subprocess.CompletedProcess(command, 0, "", "") + raise + if args[0] == "apply": + self._docker_created = True if args[0] == "get": phase = "Running" if done.stdout.startswith("true") else "Failed" + if phase == "Failed": + logs = self._run_docker(["docker", "logs", "--tail", "10", self._docker_name], timeout) + self._diagnostics.append( + {"step": "container", "exit_code": int(done.stdout.split()[-1]), "stderr": docker_diagnostic(logs.stderr)} + ) done.stdout = json.dumps({"metadata": {"name": self._docker_name}, "status": {"phase": phase}}) return done diff --git a/benchmarks/learn/engine_trace.py b/benchmarks/learn/engine_trace.py index ebb3eaa..eed4562 100644 --- a/benchmarks/learn/engine_trace.py +++ b/benchmarks/learn/engine_trace.py @@ -179,9 +179,11 @@ def make_plan( class Inputs: """Resolve spent corpora; fetch pair source commits into the host cache.""" - def __init__(self, root, cache, allow_excerpt_fallback=False): + def __init__(self, root, cache, allow_excerpt_fallback=False, *, keep_sources=False, source_cache_mb=500): self.cache = cache self.allow_excerpt_fallback = allow_excerpt_fallback + self.keep_sources = keep_sources + self.source_cache_bytes = source_cache_mb * 1024 * 1024 self.catalog_rows = defaultdict(list) for catalog in (root / "benchmarks/pairs").rglob("catalog.toml"): for item in tomllib.loads(catalog.read_text()).get("pair", []): @@ -259,8 +261,30 @@ class Inputs: raise ValueError(f"missing catalog repository for {case.name}") return repo, parent if side == "vuln" else commit + def source_path(self, repo, commit, roots): + return self.cache / "trace-sources" / digest([repo, commit, roots]) + + def prepare_source_cache(self): + cache = self.cache / "trace-sources" + cache.mkdir(parents=True, exist_ok=True) + entries = [] + for entry in cache.iterdir(): + if entry.is_dir() and not entry.is_symlink(): + size = sum(p.stat().st_size for p in entry.rglob("*") if not p.is_symlink() and p.is_file()) + entries.append((entry.stat().st_mtime_ns, entry, size)) + total = sum(size for _, _, size in entries) + for _, entry, size in sorted(entries): + if total <= self.source_cache_bytes: + break + shutil.rmtree(entry) + total -= size + free = shutil.disk_usage(cache).free + if free < 1.5 * 1024**3: + raise OSError(f"refusing source fetch: {free / 1024**3:.2f} GB free at {cache}; at least 1.5 GB required") + def fetch_source(self, repo, commit, roots): - clone = self.cache / "trace-sources" / digest([repo, commit, roots]) + self.prepare_source_cache() + clone = self.source_path(repo, commit, roots) clone.mkdir(parents=True, exist_ok=True) url = repo if repo.startswith("https://") else f"https://github.com/{repo.removesuffix('.git')}.git" env = {**os.environ, "GIT_TERMINAL_PROMPT": "0"} @@ -348,6 +372,13 @@ class Inputs: for case, side in resolved: _materialize_side(target, case, side=side) return {"mode": "excerpt-fallback", "reason": str(exc), "repo": repo, "commit": commit} + finally: + # Only the exported copy is used by executors. Drop the clone even + # on a partial fetch, timeout, failed export or excerpt fallback. + if not self.keep_sources: + clone = self.source_path(repo, commit, roots) + if clone.exists(): + shutil.rmtree(clone) verify_functions(target, rows) return {**proof, "mode": "source", "repo": repo, "commit": commit} if all(r["source"] == "dev-php" for r in rows): @@ -845,6 +876,8 @@ def main(): parser.add_argument("--examples", type=Path, help="frozen example JSONL instead of the local memory store") parser.add_argument("--memory", type=Path, default=Path.home() / "ousast-results/plane/memory") parser.add_argument("--cache", type=Path, default=Path.home() / ".cache/openultrasast") + parser.add_argument("--keep-sources", action="store_true", help="retain fetched sources after export (still subject to cache eviction)") + parser.add_argument("--source-cache-mb", type=int, default=500, help="maximum retained source MB before each fetch (default: 500)") parser.add_argument("--out", type=Path, default=DEFAULT_OUT) parser.add_argument("--image", help="default: openultrasast:dev for Docker; public 2.0.1 image for k8s/mixed") parser.add_argument("--executor", choices=("docker", "k8s", "k8s-jobs", "mixed", "queue", "ax", "mixed-ax"), default="docker") @@ -872,6 +905,8 @@ def main(): parser.add_argument("--dry-run", action="store_true") parser.add_argument("--summary", action="store_true") args = parser.parse_args() + if args.source_cache_mb < 0: + parser.error("--source-cache-mb must be nonnegative") if args.parallel is None: args.parallel = 2 if args.executor in {"ax", "mixed-ax"} else 1 if args.executor in {"ax", "mixed-ax"}: @@ -912,7 +947,7 @@ def main(): units = join_units(read_jsonl(args.units), examples, args.include_typescript) except ValueError as exc: parser.error(str(exc)) - inputs = Inputs(ROOT, args.cache, args.allow_excerpt_fallback) + inputs = Inputs(ROOT, args.cache, args.allow_excerpt_fallback, keep_sources=args.keep_sources, source_cache_mb=args.source_cache_mb) # Resolve pair source names for leak-audit coverage before calculating the resume key. for unit in units: if unit.get("excerpt") or unit["source"].startswith("advisory-fixes"): diff --git a/benchmarks/measurements/2026-10-04-unseen-04-dispatcher/record.json b/benchmarks/measurements/2026-10-04-unseen-04-dispatcher/record.json new file mode 100644 index 0000000..fa8ad56 --- /dev/null +++ b/benchmarks/measurements/2026-10-04-unseen-04-dispatcher/record.json @@ -0,0 +1,110 @@ +{ + "cleanup": "completed-before-checkpoint", + "commit": "5dc86c2f3f54b4dfb59060c88fb3ab5b307ef448", + "differences": 0, + "image": "ghcr.io/norandom/ousast-engine-task@sha256:b73c9bbd48fbf39c14ac013e5b0ff0a9d536b3d6388b50b1bd4cddfa22264b91", + "pins": [ + { + "baseline_digest": "c12c675f816a24ab33a63336a21efdc93db10a37dcf83437d91bec6f728cef1f", + "equal": true, + "instrument": [ + { + "bytes": 1761, + "cpg_bytes": 138723, + "jvm": [ + { + "exit_code": null, + "heap_mb": 1700, + "seconds": 11.88592994, + "success": null + }, + { + "exit_code": null, + "heap_mb": 1700, + "seconds": 35.821929360999995, + "success": true + }, + { + "exit_code": null, + "heap_mb": 1700, + "seconds": 7.4493522819999995, + "success": true + }, + { + "exit_code": null, + "heap_mb": 1700, + "seconds": 2.9803447330000026, + "success": true + }, + { + "exit_code": null, + "heap_mb": 1700, + "seconds": 0.5151375319999971, + "success": true + }, + { + "exit_code": null, + "heap_mb": 1700, + "seconds": 9.691500790999996, + "success": true + } + ] + } + ], + "pin_digest": "63c992b500590a9d4b643c9458df1d6cf71d770b06cc70594acbdba99b06c00b", + "record_digest": "c12c675f816a24ab33a63336a21efdc93db10a37dcf83437d91bec6f728cef1f", + "seconds": 158.85258329089265 + }, + { + "baseline_digest": "5a60ed069d99f9325a1d3f3a7612fcf0d853d4fabfd53add81151b051132a5ee", + "equal": true, + "instrument": [ + { + "bytes": 1791, + "cpg_bytes": 138886, + "jvm": [ + { + "exit_code": null, + "heap_mb": 1700, + "seconds": 7.656527941999999, + "success": null + }, + { + "exit_code": null, + "heap_mb": 1700, + "seconds": 27.532885242, + "success": true + }, + { + "exit_code": null, + "heap_mb": 1700, + "seconds": 9.524187851, + "success": true + }, + { + "exit_code": null, + "heap_mb": 1700, + "seconds": 5.0155780199999995, + "success": true + }, + { + "exit_code": null, + "heap_mb": 1700, + "seconds": 0.6809096819999993, + "success": true + }, + { + "exit_code": null, + "heap_mb": 1700, + "seconds": 13.824248304000001, + "success": true + } + ] + } + ], + "pin_digest": "8bd2f48dfb2d00cfc5b60a5dee3a8c87e6970671121caa17221f2449c1254fb2", + "record_digest": "5a60ed069d99f9325a1d3f3a7612fcf0d853d4fabfd53add81151b051132a5ee", + "seconds": 142.5083134449087 + } + ] +} diff --git a/benchmarks/unseen/README.md b/benchmarks/unseen/README.md index 9af5c38..5ca7d9d 100644 --- a/benchmarks/unseen/README.md +++ b/benchmarks/unseen/README.md @@ -289,3 +289,84 @@ proof is missing, or quick-tier findings differ. It does not convert a quick-onl without a JVM into an engine proof. Pool replays use `--lane mixed` (default) for the two AX slots and single VM overflow/fallback lane; `--arm long_deadline` changes only the hook deadline to 300 seconds and is a separate input digest. + +### Deadline calibration and native agreement + +The 2026-10-05 maintainer decision supersedes the unscaled cluster deadline above. +gVisor snapshot preparation can consume 30 seconds before any analysis. Calibrate +the pinned replay image against the VM's native Docker **runc** runtime first. +The default deterministic sample is the first 20 changes ordered by SHA-256 of +change ID; a smaller manifest (including the five-change proof) uses all changes. +Sampling never depends on manifest order. + +`calibrate` runs both lanes with a 600-second hook deadline (override `--deadline` +if stages still time out). It reads resolution, provenance, preparation, snapshot, +and quick timings from the push artifact; snapshot is nested in preparation. +`quick_total` sums resolution + provenance + preparation + quick, without counting +snapshot twice. Engine build and query times sum across the hook's scans. Hook +wall time comes from the instrument block. Missing/nonpositive timings, deadline +exhaustion, degraded engine scans or failed instruments prevent a successful +calibration; failed changes are never silently dropped. Each factor is the median +of paired cluster/native ratios with a deterministic 2,000-resample percentile +bootstrap 95% interval. The measurement record contains opaque change digests, +counts and timings, with no repository names. Raw artifacts remain outside the repo. + +Set `REPLAY_IMAGE` to the newly built replay image's pinned digest and +`REPLAY_IMAGE_BYTES` to an **unpacked image size upper bound in bytes** from the +image build. Do not use compressed registry layer size. Docker refuses to pull +without this bound or when `df` at its data root reports less than that bound plus +1 GiB free. It then explicitly pulls the image before issuing presigned task URLs. +Supply `--docker-image-bytes "$REPLAY_IMAGE_BYTES"` to VM and mixed replay commands, +including the earlier VM proof command. +Launch/pull failures retain exit codes and the last ten sanitized stderr lines in +`docker-error.json`, the result record and attempt history. URLs, known environment +secrets and credential assignments are redacted. The original VM failure's cause +remains unconfirmed until this diagnostic is captured; the old transport discarded it. + +Coordinator commands (these run networked workloads; offline tests use fakes): + +```sh +export PATH=/home/mc/Source/OpenUltraSAST/.venv/bin:$PATH +export PYTHONPATH=$PWD/src +export S3_ENDPOINT=https://files.because-security.com +CALIBRATION_RECORD="benchmarks/measurements/$(date +%F)-unseen-calibration/record.json" +python -m benchmarks.unseen.calibrate \ + --manifest "$HOME/ousast-results/unseen/proof-05-inputs.json" \ + --sample 20 --deadline 600 --image "$REPLAY_IMAGE" \ + --docker-image-bytes "$REPLAY_IMAGE_BYTES" --kubeconfig "$AX_KUBECONFIG" \ + --out "$HOME/ousast-results/unseen/calibration" --record "$CALIBRATION_RECORD" + +python -m benchmarks.unseen.replay --manifest benchmarks/unseen/pool-p1.toml \ + --pool p1 --slice 1 --arm default --lane ax --image "$REPLAY_IMAGE" \ + --calibration "$CALIBRATION_RECORD" --native-sample 100 \ + --docker-image-bytes "$REPLAY_IMAGE_BYTES" --kubeconfig "$AX_KUBECONFIG" \ + --out "$HOME/ousast-results/unseen/p1/1/calibrated-default" +python -m benchmarks.unseen.score "$HOME/ousast-results/unseen/p1/1/calibrated-default" \ + --out "$HOME/ousast-results/unseen/p1/1/calibrated-default-score.json" +``` + +For a 20-change slice calibration, replace the proof manifest with +`benchmarks/unseen/pool-p1.toml --slice 1`; both commands accept `--private-manifest`. +For the long arm use `--arm long_deadline` and a separate output root. Calibration +requires the same image digest as replay. An explicit `--deadline-scale FACTOR` +is an alternative to `--calibration`, not an additional multiplier. + +Cluster deadlines are `round(30 * factor)` or `round(300 * factor)` seconds; +native deadlines remain 30 or 300, including VM fallback in mixed runs. +Records carry configured and applied scale, base/applied deadline and actual lane; +the scorer reports their counts. Task, subprocess and presigned-object budgets +grow with the applied deadline. + +Use `--lane ax` (or `kind`) with `--native-sample K` for an agreement run, so every +sampled change has a cluster counterpart. The native sample lives in `OUT/native`; +the scorer detects it automatically, or accepts `--native PATH`. Missing, failed, +wrong-lane or mismatched-image/arm pairs are unknown, never quiet agreements. +The decision is whether any of the three finding streams flags the push, not +whether the finding matches the vulnerability label. + +**Decision rule:** if the Wilson 95% agreement lower bound is below 95%, do not use +the cluster results for the baseline. The score sets `flagged=true`, +`baseline_eligible=false` and the adoption gate to `not_run`. Unknown pairs also +flag the run. Disagreements are listed by change ID only, alongside per-change +flagged/not-flagged decisions. Even perfect agreement needs at least 73 pairs to +clear this bound; the five-change proof or a 20-change sample cannot qualify it. diff --git a/benchmarks/unseen/calibrate.py b/benchmarks/unseen/calibrate.py new file mode 100644 index 0000000..2bf9d83 --- /dev/null +++ b/benchmarks/unseen/calibrate.py @@ -0,0 +1,128 @@ +"""Paired native/cluster deadline calibration; publish only opaque IDs and numbers.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import random +import re +import statistics +from datetime import date +from pathlib import Path + +from benchmarks.ax.batch import write_json +from benchmarks.unseen.replay import canonical, load_slice, run_replays, sample_changes +from benchmarks.unseen.score import coverage, load_records + + +def stage_times(record): + if coverage(record) == "instrument_failure": + raise ValueError("calibration instrument failed") + push = record["push"] + if "deadline_exhausted" in json.dumps(push) or any(t.get("status") == "failed" for t in push.get("quick_tier", [])): + raise ValueError("calibration deadline exhausted or quick tier failed; increase deadline") + raw = push.get("timings", {}) + timings = {} + for key, value in raw.items(): + key = re.sub(r"^comparison_\d+_", "", key) + if not isinstance(value, (int, float)) or not math.isfinite(value) or value < 0: + raise ValueError("invalid stage timing") + timings[key] = timings.get(key, 0) + value + required = ("resolution_seconds", "provenance_seconds", "preparation_seconds", "quick_seconds", "snapshot_seconds") + if any(key not in timings for key in required): + raise ValueError("missing completed quick-stage timings") + scans = [r["scan"] for r in push.get("scans", [])] + if any(s.get("degradations") for s in scans): + raise ValueError("incomplete engine stages; calibration cannot use degraded scans") + if not scans or any(k not in s for s in scans for k in ("build_seconds", "query_seconds")): + raise ValueError("missing engine timings") + result = { + "snapshot": timings["snapshot_seconds"], # Nested inside preparation; never counted twice. + "preparation": timings["preparation_seconds"], + "quick": timings["quick_seconds"], + "quick_total": sum(timings[k] for k in required if k != "snapshot_seconds"), + "engine_build": sum(s["build_seconds"] for s in scans), + "engine_queries": sum(s["query_seconds"] for s in scans), + "hook": record["instrument"].get("timings", {}).get("pre-push"), + } + if any(not isinstance(v, (int, float)) or not math.isfinite(v) or v <= 0 for v in result.values()): + raise ValueError("missing or nonpositive stage timing") + return result + + +def slowdown(native, cluster, *, seed=0, draws=2000): + if not native or len(native) != len(cluster): + raise ValueError("calibration needs paired timings") + if any(not math.isfinite(v) or v <= 0 for v in [*native, *cluster]): + raise ValueError("timings must be finite and positive") + ratios = [b / a for a, b in zip(native, cluster, strict=True)] + rng = random.Random(seed) + bootstrap = sorted(statistics.median(rng.choices(ratios, k=len(ratios))) for _ in range(draws)) + return { + "n": len(ratios), + "factor": statistics.median(ratios), + "bootstrap95": [bootstrap[int(0.025 * (draws - 1))], bootstrap[int(0.975 * (draws - 1))]], + } + + +def calibration_record(native, cluster): + nn, cc = ({r["id"]: r for r in rows} for rows in (native, cluster)) + if not nn or nn.keys() != cc.keys() or len(nn) != len(native) or len(cc) != len(cluster): + raise ValueError("calibration requires exactly paired unique changes") + pairs = [] + for key in sorted(nn): + if nn[key].get("lane") != "docker" or cc[key].get("lane") != "ax": + raise ValueError("calibration requires native docker and cluster ax lanes") + pairs.append( + {"change_id": hashlib.sha256(str(key).encode()).hexdigest(), "native": stage_times(nn[key]), "cluster": stage_times(cc[key])} + ) + return { + "status": "ok", + "changes": len(pairs), + "pairs": pairs, + **{ + stage: slowdown([p["native"][stage] for p in pairs], [p["cluster"][stage] for p in pairs]) + for stage in ("quick_total", "engine_build", "engine_queries") + }, + } + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--manifest", type=Path, required=True) + parser.add_argument("--private-manifest", type=Path) + parser.add_argument("--slice", default="1") + parser.add_argument("--sample", type=int, default=20) + parser.add_argument("--deadline", type=int, default=600) + parser.add_argument("--image", required=True) + parser.add_argument("--docker-image-bytes", type=int, required=True, help="unpacked image size upper bound") + parser.add_argument("--ax-bin", default="ax") + parser.add_argument("--kubeconfig", type=Path) + parser.add_argument("--atespace", default="default") + parser.add_argument("--out", type=Path, default=Path.home() / "ousast-results/unseen/calibration") + parser.add_argument("--record", type=Path, default=Path("benchmarks/measurements") / f"{date.today()}-unseen-calibration/record.json") + args = parser.parse_args() + if args.deadline < 1 or args.sample < 1: + parser.error("deadline and sample must be positive") + args.arm = "default" + changes = sample_changes(load_slice(args.manifest, args.slice, args.private_manifest), args.sample) + failed = False + for lane, directory in (("docker", "native"), ("ax", "cluster")): + failed |= run_replays(changes, args, args.out / directory, lane, deadline=args.deadline) + result = {"status": "failed", "changes": len(changes)} + if not failed: + try: + result = calibration_record(load_records(args.out / "native"), load_records(args.out / "cluster")) + except ValueError as exc: + result["reason"] = str(exc) # Only locally controlled diagnostics. + result.update( + image=args.image, deadline_seconds=args.deadline, sample_digest=hashlib.sha256(canonical([c["id"] for c in changes])).hexdigest() + ) + write_json(args.record, result) + return 0 if result["status"] == "ok" else 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/unseen/proof.py b/benchmarks/unseen/proof.py index b5f2fb0..6effa0e 100644 --- a/benchmarks/unseen/proof.py +++ b/benchmarks/unseen/proof.py @@ -57,7 +57,7 @@ def replay_proof(roots): def engine_proof(root, baseline): current = records(root) old = {digest([r["repo"], r["pin"]]): r for r in records(baseline) if "repo" in r and "pin" in r} - current = [r for r in current if "repo" in r and "pin" in r] + current = [r for r in current if "repo" in r and "pin" in r and not all(u.get("status") == "unsupported" for u in r["units"])] if len(current) != 2: raise ValueError("engine proof requires exactly two pins") evidence = [] diff --git a/benchmarks/unseen/replay.py b/benchmarks/unseen/replay.py index ad1de0c..b5830e5 100644 --- a/benchmarks/unseen/replay.py +++ b/benchmarks/unseen/replay.py @@ -6,6 +6,7 @@ import argparse import hashlib import io import json +import math import subprocess import tarfile import tomllib @@ -15,11 +16,26 @@ from benchmarks.ax.batch import AXLane, DockerLane, Item, Workload, dispatch, wr from benchmarks.unseen.score import coverage -def make_item(change, arm, image): +def make_item(change, arm, image, *, deadline_scale=1.0, lane="ax", deadline=None): if arm not in {"default", "long_deadline"}: raise ValueError("unknown replay arm") identity = hashlib.sha256(str(change["id"]).encode()).hexdigest()[:14] - spec = {**change, "id": identity, "hook_flags": [] if arm == "default" else ["--deadline", "300"], "image": image} + if not math.isfinite(deadline_scale) or deadline_scale <= 0: + raise ValueError("deadline scale must be finite and positive") + base = deadline if deadline is not None else 30 if arm == "default" else 300 + applied = round(base * (1 if lane == "docker" else deadline_scale)) + if applied < 1: + raise ValueError("scaled deadline must be at least one second") + spec = { + **change, + "id": identity, + "hook_flags": ["--deadline", str(applied)], + "image": image, + "base_deadline_seconds": base, + "deadline_seconds": applied, + "deadline_scale": deadline_scale, + "applied_deadline_scale": 1.0 if lane == "docker" else deadline_scale, + } data = json.dumps(spec, sort_keys=True).encode() digest = hashlib.sha256(data).hexdigest() return Item( @@ -146,6 +162,11 @@ def main(): parser.add_argument("--slice", default="1") parser.add_argument("--out", type=Path) parser.add_argument("--lane", choices=("ax", "kind", "docker", "mixed"), default="mixed") + group = parser.add_mutually_exclusive_group() + group.add_argument("--deadline-scale", type=float, default=1.0) + group.add_argument("--calibration", type=Path) + parser.add_argument("--native-sample", type=int, default=0) + parser.add_argument("--docker-image-bytes", type=int, help="unpacked image size upper bound, required for VM pull") parser.add_argument("--ax-bin", default="ax") parser.add_argument("--kubeconfig", type=Path) parser.add_argument("--atespace", default="default") @@ -157,7 +178,15 @@ def main(): if not args.image: parser.error("--image digest is required") changes = load_slice(args.manifest, args.slice, args.private_manifest) - items = [make_item(change, args.arm, args.image) for change in changes] + factor = args.deadline_scale + if args.calibration: + calibration = json.loads(args.calibration.read_text()) + if calibration.get("status") != "ok" or calibration.get("image") != args.image: + parser.error("calibration must be successful and match the replay image") + factor = calibration["quick_total"]["factor"] + if args.native_sample < 0 or (args.native_sample and args.lane not in {"ax", "kind"}): + parser.error("--native-sample requires an ax or kind cluster run and a positive K") + items = [make_item(change, args.arm, args.image, deadline_scale=factor, lane=args.lane) for change in changes] if len({i.id for i in items}) != len(items): parser.error("duplicate change identity") args.out = args.out or Path.home() / "ousast-results/unseen" / args.pool / args.slice / args.arm @@ -166,27 +195,52 @@ def main(): return 0 if args.out.resolve().is_relative_to(Path(__file__).resolve().parents[2]): parser.error("raw replay output must be outside the repository") + failed = run_replays(changes, args, args.out, args.lane, deadline_scale=factor) + if args.native_sample: + failed |= run_replays(sample_changes(changes, args.native_sample), args, args.out / "native", "docker", deadline_scale=factor) + return 2 if failed else 0 + + +def sample_changes(changes, count): + if count < 1: + raise ValueError("sample size must be positive") + if len({str(c["id"]) for c in changes}) != len(changes): + raise ValueError("duplicate change identity") + return sorted(changes, key=lambda c: hashlib.sha256(str(c["id"]).encode()).digest())[:count] + + +def run_replays(changes, args, out, lane, *, deadline_scale=1.0, deadline=None): + if out.resolve().is_relative_to(Path(__file__).resolve().parents[2]): + raise ValueError("raw replay output must be outside the repository") + items = [make_item(c, args.arm, args.image, deadline_scale=deadline_scale, lane=lane, deadline=deadline) for c in changes] write_json( - args.out / "inputs.json", + out / "inputs.json", [{"item_id": item.id, "change": change, "input_digest": item.input_digest} for item, change in zip(items, changes, strict=True)], ) - workload = Workload(args.image, frozenset({"SPEC_URL", "RESULT_URL"}), 600 if args.arm == "default" else 900, validate_result) - lanes = ("ax", "ax", "docker") if args.lane == "mixed" else (args.lane,) * (2 if args.lane in {"ax", "kind"} else 1) + maximum = max( + max(json.loads(i.inputs["SPEC_URL"])["deadline_seconds"], json.loads(i.inputs["SPEC_URL"])["base_deadline_seconds"]) for i in items + ) + workload = Workload(args.image, frozenset({"SPEC_URL", "RESULT_URL"}), maximum + 600, validate_result) + lanes = ("ax", "ax", "docker") if lane == "mixed" else (lane,) * (2 if lane in {"ax", "kind"} else 1) + original = {item.id: change for item, change in zip(items, changes, strict=True)} executors = {} - for lane in set(lanes): - transport = (DockerLane if lane == "docker" else AXLane)(args, workload) + for name in set(lanes): + transport = (DockerLane if name == "docker" else AXLane)(args, workload) - def execute(item, output, attempt, transport=transport): - record = {**json.loads(item.inputs["SPEC_URL"]), **transport(item, output, attempt)} + def execute(item, output, attempt, transport=transport, name=name): + spec = original[item.id] + # Re-render on placement, including VM fallback: native is always unscaled. + placed = make_item(spec, args.arm, args.image, deadline_scale=deadline_scale, lane=name, deadline=deadline) + placed.id = item.id + record = {**json.loads(placed.inputs["SPEC_URL"]), "lane": name, **transport(placed, output, attempt)} if (output / "result.dat").exists(): - # Durable archive lives separately from attempt staging objects, which are deleted. - key = f"unseen/{item.input_digest}/result.dat" + key = f"unseen/{placed.input_digest}/{name}/result.dat" transport.store._put(key, (output / "result.dat").read_bytes()) record["result_object"] = key return record - executors[lane] = execute - return 2 if dispatch(items, workload, args.out, executors, lanes=lanes) else 0 + executors[name] = execute + return dispatch(items, workload, out, executors, lanes=lanes) if __name__ == "__main__": diff --git a/benchmarks/unseen/replay_entry.py b/benchmarks/unseen/replay_entry.py index 09f9a34..7dcdef9 100644 --- a/benchmarks/unseen/replay_entry.py +++ b/benchmarks/unseen/replay_entry.py @@ -52,7 +52,8 @@ def validate_spec(spec): raise ReplayFailure("repository must use public DNS") if any(not re.fullmatch("[0-9a-f]{40,64}", spec[key]) for key in ("base", "head")): raise ReplayFailure("invalid revision") - if spec.get("hook_flags", []) not in ([], ["--deadline", "300"]): + flags = spec.get("hook_flags", []) + if flags and (len(flags) != 2 or flags[0] != "--deadline" or not re.fullmatch(r"[1-9][0-9]*", str(flags[1]))): raise ReplayFailure("unsupported hook flags") @@ -92,6 +93,8 @@ def main(*, work_root=Path("/tmp"), runner=subprocess.run, heap_profile="cluster spec = json.loads(spec_path.read_text()) instrument["image"] = spec.get("image") validate_spec(spec) + instrument["deadline_seconds"] = int(spec["hook_flags"][1]) if spec.get("hook_flags") else 30 + instrument["deadline_scale"] = spec.get("deadline_scale", 1.0) case = work_root / "case" case.mkdir() git = ["git", "-c", "core.hooksPath=/dev/null", "-c", "protocol.file.allow=never", "-C", str(case)] @@ -148,7 +151,9 @@ def main(*, work_root=Path("/tmp"), runner=subprocess.run, heap_profile="cluster *spec.get("hook_flags", []), ] before = clock() - completed = runner(command, capture_output=True, text=True, timeout=330 if spec.get("hook_flags") else 60, env=env) + completed = runner( + command, capture_output=True, text=True, timeout=int(spec["hook_flags"][1]) + 30 if spec.get("hook_flags") else 60, env=env + ) instrument["timings"][step] = clock() - before instrument["hook_exit"] = completed.returncode (output / "push.txt").write_text(completed.stdout) diff --git a/benchmarks/unseen/score.py b/benchmarks/unseen/score.py index f0116bf..bc1932c 100644 --- a/benchmarks/unseen/score.py +++ b/benchmarks/unseen/score.py @@ -199,12 +199,59 @@ def score(records, *, seed=0): return { **metrics(usable), "total": len(records), + "deadlines": [ + {"lane": lane, "deadline_seconds": deadline, "deadline_scale": factor, "applied_deadline_scale": applied, "count": count} + for (lane, deadline, factor, applied), count in sorted( + Counter( + (r.get("lane", "unknown"), r.get("deadline_seconds"), r.get("deadline_scale"), r.get("applied_deadline_scale")) + for r in records + ).items(), + key=lambda pair: str(pair[0]), + ) + ], "coverage": {state: counts[state] for state in STATES}, "breakdowns": breakdown, "confirmed": metrics([r for r in usable if r.get("label_check", r.get("confirmed")) in ("confirmed", True)]), } +def agreement(cluster, native): + """Missing, failed or wrong-lane records are unknown, never matching quiet pushes.""" + cc = {r["id"]: r for r in cluster} + if len(cc) != len(cluster) or len({r["id"] for r in native}) != len(native): + raise ValueError("duplicate agreement identities") + pairs, unknown, disagreements = [], [], [] + for n in sorted(native, key=lambda r: r["id"]): + c = cc.get(n["id"]) + if ( + c is None + or c.get("lane") not in {"ax", "kind"} + or n.get("lane") != "docker" + or c.get("image") != n.get("image") + or c.get("base_deadline_seconds") != n.get("base_deadline_seconds") + or coverage(c) == "instrument_failure" + or coverage(n) == "instrument_failure" + ): + unknown.append(n["id"]) + continue + cluster_flag, native_flag = false_alarm(c), false_alarm(n) + same = cluster_flag == native_flag + pairs.append({"change_id": n["id"], "cluster_flagged": cluster_flag, "native_flagged": native_flag, "same": same}) + if not same: + disagreements.append(n["id"]) + report = rate([p["same"] for p in pairs]) + eligible = bool(pairs) and not unknown and report["wilson95"][0] >= 0.95 + return { + **report, + "requested": len(native), + "pairs": pairs, + "unknown": unknown, + "disagreements": disagreements, + "baseline_eligible": eligible, + "flagged": not eligible, + } + + def budgets(rows, *, thresholds=None, binary=False): ordinary = [r for r in rows if r["kind"] == "ordinary"] vulnerable = [r for r in rows if r["kind"] != "ordinary"] @@ -295,12 +342,20 @@ def main(): parser.add_argument("records", type=Path, help="JSON array of joined change/instrument/push records") parser.add_argument("--out", type=Path, required=True) parser.add_argument("--against", type=Path, help="production-arm records for a paired adoption gate") + parser.add_argument("--native", type=Path, help="native sample root; auto-detected under replay root") parser.add_argument("--slice-id") parser.add_argument("--informed", action="store_true") args = parser.parse_args() current = load_records(args.records) result = score(current) + native = args.native or (args.records / "native" if args.records.is_dir() and (args.records / "native").exists() else None) + if native: + result["native_agreement"] = agreement(current, load_records(native)) + result["baseline_eligible"] = result["native_agreement"]["baseline_eligible"] + result["flagged"] = result["native_agreement"]["flagged"] result["unseen"] = gate(load_records(args.against) if args.against else [], current, slice_id=args.slice_id, informed=args.informed) + if native and result["flagged"]: + result["unseen"] = {"gate": "not_run", "reason": "native_agreement"} args.out.write_text(json.dumps(result, sort_keys=True, indent=2) + "\n") diff --git a/tests/test_engine_trace.py b/tests/test_engine_trace.py index b46a9af..722e9c2 100644 --- a/tests/test_engine_trace.py +++ b/tests/test_engine_trace.py @@ -1719,3 +1719,104 @@ def test_worker_session_question_deadline_keeps_markers_and_does_not_replay(tmp_ assert backend.timed_out and backend.asked == ["a"] assert not backend.answers and not backend.session_fallback assert closed == [True] + + +@pytest.mark.parametrize("keep_sources", [False, True]) +@pytest.mark.parametrize("outcome", ["success", "failure", "timeout", "fetch-failure", "fetch-timeout", "export-failure"]) +def test_finished_pin_source_lifetime(tmp_path, monkeypatch, keep_sources, outcome): + inputs, case = pair_inputs(tmp_path) + inputs.keep_sources = keep_sources + pin = trace.make_plan([unit(file=case.relpath)], tmp_path)[0] + clone = inputs.source_path(case.repo, "a" * 40, ["pkg"]) + monkeypatch.setattr(inputs, "pair_for", lambda row: (case, "vuln")) + + def git(command, **kwargs): + (clone / "pkg").mkdir(exist_ok=True) + (clone / case.relpath).write_text("def run(): pass\n") + if outcome == "fetch-failure": + raise subprocess.CalledProcessError(1, command) + if outcome == "fetch-timeout": + raise subprocess.TimeoutExpired(command, 300) + if outcome == "export-failure": + (clone / case.relpath).unlink() + return subprocess.CompletedProcess(command, 0, "a" * 40 + "\n", "") + + class Dispatcher: + def execute(self, checkout, output, prepared, name): + assert (checkout / case.relpath).read_text() == "def run(): pass\n" + if outcome == "timeout": + return None, "worker" + if outcome == "failure": + raise OSError("executor failed") + trace.write_json( + output / "result.json", + {**prepared, "done": True, "units": [worker.unit_record(u, "asked-nothing") for u in prepared["units"]]}, + ) + return subprocess.CompletedProcess([], 0, "", ""), "worker" + + monkeypatch.setattr(trace.subprocess, "run", git) + monkeypatch.setattr(trace, "prepare_questions", lambda *_: {"a": {}}) + args = k8s_args(tmp_path, executor="ax", ax_dispatcher=Dispatcher()) + assert trace.run_pin(pin, None, {}, inputs, tmp_path, args) is (outcome != "success") + assert clone.exists() is keep_sources + + +def test_source_cache_evicts_oldest_before_fetch(tmp_path, monkeypatch): + from types import SimpleNamespace + + inputs, _ = pair_inputs(tmp_path) + inputs.source_cache_bytes = 8 + cache = inputs.cache / "trace-sources" + for name, modified in (("oldest", 10), ("newest", 30), ("middle", 20)): + entry = cache / name + entry.mkdir(parents=True) + (entry / "data").write_bytes(b"123456") + trace.os.utime(entry, (modified, modified)) + # Other caches and symlink targets must remain untouched. + other = inputs.cache / "repos" + other.mkdir() + (other / "data").write_bytes(b"keep") + (cache / "link").symlink_to(other, target_is_directory=True) + monkeypatch.setattr(trace.shutil, "disk_usage", lambda path: SimpleNamespace(free=2 * 1024**3)) + commands = [] + + def git(command, **kwargs): + assert not (cache / "oldest").exists() + assert not (cache / "middle").exists() + assert (cache / "newest/data").read_bytes() == b"123456" + assert (other / "data").read_bytes() == b"keep" + commands.append(command) + return subprocess.CompletedProcess(command, 0, "a" * 40 + "\n", "") + + monkeypatch.setattr(trace.subprocess, "run", git) + inputs.fetch_source("owner/repo", "a" * 40, ["pkg"]) + assert any("fetch" in command for command in commands) + + +def test_source_fetch_refuses_low_disk_before_git(tmp_path, monkeypatch): + from types import SimpleNamespace + + inputs, _ = pair_inputs(tmp_path) + monkeypatch.setattr(trace.shutil, "disk_usage", lambda path: SimpleNamespace(free=int(1.5 * 1024**3) - 1)) + monkeypatch.setattr(trace.subprocess, "run", lambda *a, **kw: pytest.fail("low disk must refuse before git")) + with pytest.raises(OSError, match="refusing source fetch: .*at least 1.5 GB required"): + inputs.fetch_source("owner/repo", "a" * 40, ["pkg"]) + assert not inputs.source_path("owner/repo", "a" * 40, ["pkg"]).exists() + + +@pytest.mark.parametrize("flags,keep,cap", [([], False, 500), (["--keep-sources", "--source-cache-mb", "7"], True, 7)]) +def test_source_cache_cli_options(tmp_path, monkeypatch, flags, keep, cap): + original = trace.Inputs + created = [] + + def inputs(*args, **kwargs): + result = original(*args, **kwargs) + created.append(result) + return result + + monkeypatch.setattr(trace, "Inputs", inputs) + monkeypatch.setattr(trace, "read_jsonl", lambda path: []) + monkeypatch.setattr(sys, "argv", [str(SCRIPT), "--dry-run", "--examples", "unused", "--cache", str(tmp_path), *flags]) + assert trace.main() == 0 + assert created[0].keep_sources is keep + assert created[0].source_cache_bytes == cap * 1024 * 1024 diff --git a/tests/test_unseen_calibrate.py b/tests/test_unseen_calibrate.py new file mode 100644 index 0000000..f1df66c --- /dev/null +++ b/tests/test_unseen_calibrate.py @@ -0,0 +1,235 @@ +"""Offline calibration, placement, agreement and Docker diagnostic controls.""" + +import importlib +import json +import subprocess +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +c = importlib.import_module("benchmarks.unseen.calibrate") +r = importlib.import_module("benchmarks.unseen.replay") +s = importlib.import_module("benchmarks.unseen.score") +b = importlib.import_module("benchmarks.ax.batch") + + +def timing_record(identity="opaque", factor=1, lane="docker"): + return { + "id": identity, + "lane": lane, + "repo": "never-publish", + "kind": "ordinary", + "instrument": { + "head_verified": True, + "base_verified": True, + "changed_bytes": {"app.py": 10}, + "hook_exit": 0, + "hook_bytes_read": 10, + "timings": {"pre-push": 60 * factor}, + }, + "push": { + "timings": { + f"comparison_0_{k}_seconds": v * factor + for k, v in (("resolution", 1), ("provenance", 2), ("snapshot", 4), ("preparation", 6), ("quick", 1)) + }, + "quick_tier": [{"bytes_read": 10, "findings": []}], + "scans": [{"scan": {"build_seconds": 20 * factor, "query_seconds": 30 * factor}}], + }, + } + + +def test_calibration_median_bootstrap_and_privacy(): + result = c.slowdown([1, 10, 100], [2, 20, 200]) + assert result == {"n": 3, "factor": 2, "bootstrap95": [2, 2]} + varied = c.slowdown([1] * 5, [1, 2, 3, 4, 5]) + assert varied["factor"] == 3 + assert varied["bootstrap95"] == [1, 5] + assert varied == c.slowdown([1] * 5, [1, 2, 3, 4, 5]) + record = c.calibration_record([timing_record()], [timing_record(factor=3, lane="ax")]) + assert record["quick_total"]["factor"] == 3 + assert record["pairs"][0]["native"]["quick_total"] == 10 # Snapshot is nested, not added twice. + assert record["engine_build"]["factor"] == record["engine_queries"]["factor"] == 3 + assert "never-publish" not in json.dumps(record) + with pytest.raises(ValueError): + c.slowdown([0], [2]) + bad = timing_record() + bad["push"]["snapshots"] = [{"boundaries": [{"reason": "deadline_exhausted"}]}] + with pytest.raises(ValueError, match="deadline"): + c.stage_times(bad) + bad = timing_record() + del bad["push"]["timings"]["comparison_0_quick_seconds"] + with pytest.raises(ValueError, match="missing"): + c.stage_times(bad) + + +@pytest.mark.parametrize( + "lane,arm,expected", + [ + ("ax", "default", 75), + ("ax", "long_deadline", 750), + ("docker", "default", 30), + ("docker", "long_deadline", 300), + ], +) +def test_scaled_task_contract(lane, arm, expected): + item = r.make_item({"id": "a"}, arm, "image", deadline_scale=2.5, lane=lane) + spec = json.loads(item.inputs["SPEC_URL"]) + assert spec["hook_flags"] == ["--deadline", str(expected)] + assert spec["deadline_seconds"] == expected and spec["deadline_scale"] == 2.5 + entry = importlib.import_module("benchmarks.unseen.replay_entry") + entry.validate_spec({**spec, "repository_url": "https://public.example/test", "base": "a" * 40, "head": "b" * 40}) + + +def test_native_sample_and_agreement(): + native = [timing_record(str(i)) for i in range(100)] + cluster = [timing_record(str(i), lane="ax") for i in range(100)] + assert r.sample_changes(native, 20) == r.sample_changes(list(reversed(native)), 20) + report = s.agreement(cluster, native) + assert report["rate"] == 1 and report["baseline_eligible"] + assert s.agreement(cluster[:20], native[:20])["flagged"] # Perfect small sample is insufficient. + cluster[0]["push"]["quick_tier"][0]["findings"] = [{"cwe": "CWE-89"}] + report = s.agreement(cluster, native) + assert report["disagreements"] == ["0"] and report["rate"] == 0.99 + assert report["flagged"] + cluster[0]["instrument"] = {} + report = s.agreement(cluster, native) + assert report["unknown"] == ["0"] and report["flagged"] + assert s.agreement([], [])["flagged"] + + +def test_docker_error_capture_and_preflight(tmp_path, monkeypatch): + calls = [] + monkeypatch.setenv("TEST_SECRET", "hidden-value") + lane = object.__new__(b.DockerLane) + lane.args = SimpleNamespace(docker_image_bytes=100) + lane.workload = SimpleNamespace(image="pinned-image") + lane._prepared = False + + def runner(command, **kwargs): + calls.append(command) + if command[:2] == ["docker", "info"]: + return subprocess.CompletedProcess(command, 0, "/docker-data\n", "") + if command[0] == "df": + return subprocess.CompletedProcess(command, 0, "Avail\n2000000000\n", "") + return subprocess.CompletedProcess(command, 125, "", "pull denied hidden-value https://host.example/?signature=private\nlast error") + + lane.runner = runner + record = lane(None, tmp_path, 0) + assert record["status"] == "instrument_failure" + assert record["docker_errors"][0]["exit_code"] == 125 + assert record["docker_errors"][0]["stderr"].endswith("last error") + assert "hidden-value" not in json.dumps(record) and "private" not in json.dumps(record) + assert [cmd[0] for cmd in calls] == ["docker", "df", "docker"] + assert calls[-1] == ["docker", "pull", "pinned-image"] + assert (tmp_path / "docker-error.json").exists() + lane.args.docker_image_bytes = 2000000000 + calls.clear() + record = lane(None, tmp_path, 1) + assert "refusing Docker pull" in record["docker_errors"][0]["stderr"] + assert len(calls) == 2 + + +def test_docker_launch_error_survives_transport(tmp_path, monkeypatch): + monkeypatch.setenv("S3_ENDPOINT", "https://store.example") + + class Store: + def _put(self, *args): + pass + + def _delete(self, *args): + pass + + def presign_get(self, *args): + return "https://store.example/secret" + + presign_put = presign_get + + image = "registry/image@sha256:" + "a" * 64 + lane = b.DockerLane( + SimpleNamespace(atespace="default"), + b.Workload(image, frozenset({"SPEC_URL", "RESULT_URL"}), 60, None), + store=Store(), + runner=lambda command, **kw: subprocess.CompletedProcess(command, 0 if command[1] == "rm" else 125, "", "runtime missing"), + ) + lane._prepared = True + item = r.make_item({"id": "x"}, "default", image) + record = lane(item, tmp_path, 0) + assert record["docker_errors"] == [{"step": "run", "exit_code": 125, "stderr": "runtime missing"}] + assert record["status"] == "instrument_failure" + + +def test_placement_scales_cluster_but_not_vm_fallback(tmp_path, monkeypatch): + observed = [] + + class Transport: + def __init__(self, args, workload): + assert workload.deadline >= 750 + 600 + + def __call__(self, item, output, attempt): + spec = json.loads(item.inputs["SPEC_URL"]) + observed.append(spec) + return {"status": "quick_only"} + + def dispatch(items, workload, out, executors, **kwargs): + for lane in ("ax", "docker"): + record = executors[lane](items[0], out, 0) + assert record["lane"] == lane + assert record["deadline_seconds"] == (750 if lane == "ax" else 300) + return False + + monkeypatch.setattr(r, "AXLane", Transport) + monkeypatch.setattr(r, "DockerLane", Transport) + monkeypatch.setattr(r, "dispatch", dispatch) + args = SimpleNamespace(arm="long_deadline", image="test-image") + assert not r.run_replays([{"id": "opaque"}], args, tmp_path, "mixed", deadline_scale=2.5) + assert observed[0]["id"] == observed[1]["id"] + assert observed[0]["hook_flags"] == ["--deadline", "750"] + assert observed[1]["hook_flags"] == ["--deadline", "300"] + + +def test_calibration_cli_writes_anonymous_record(tmp_path, monkeypatch): + manifest, record = tmp_path / "manifest.json", tmp_path / "record.json" + changes = [{"id": str(i), "repo": "do-not-publish"} for i in range(25)] + monkeypatch.setattr(c, "load_slice", lambda *args: changes) + calls = [] + + def replay(selected, args, out, lane, **kwargs): + calls.append((selected, lane, kwargs)) + return False + + monkeypatch.setattr(c, "run_replays", replay) + monkeypatch.setattr( + c, + "load_records", + lambda path: [ + timing_record(str(i), factor=3 if path.name == "cluster" else 1, lane="ax" if path.name == "cluster" else "docker") + for i in range(20) + ], + ) + monkeypatch.setattr( + sys, + "argv", + [ + "calibrate", + "--manifest", + str(manifest), + "--image", + "digest", + "--docker-image-bytes", + "100", + "--out", + str(tmp_path / "raw"), + "--record", + str(record), + ], + ) + assert c.main() == 0 + assert len(calls[0][0]) == 20 and calls[0][0] == calls[1][0] + assert [call[1] for call in calls] == ["docker", "ax"] + assert all(call[2]["deadline"] == 600 for call in calls) + payload = record.read_text() + assert "do-not-publish" not in payload and "never-publish" not in payload + assert json.loads(payload)["quick_total"]["factor"] == 3 diff --git a/tests/test_unseen_proof.py b/tests/test_unseen_proof.py index 0f28edf..5b759f5 100644 --- a/tests/test_unseen_proof.py +++ b/tests/test_unseen_proof.py @@ -54,3 +54,33 @@ def test_engine_proof_counts_differences_without_names(tmp_path): (baseline / f"{i}.json").write_text(json.dumps(record)) proof = p.engine_proof(current, baseline) assert proof["differences"] == 0 and "synthetic" not in json.dumps(proof) + + +@pytest.mark.parametrize("executed", [2, 3]) +def test_engine_proof_counts_only_executed_records(tmp_path, executed): + current, baseline = tmp_path / "current", tmp_path / "baseline" + current.mkdir() + baseline.mkdir() + (current / "progress.json").write_text(json.dumps({"completed": executed})) + for i in range(executed + 4): + record = { + "repo": "synthetic", + "pin": str(i), + "done": True, + "seconds": 20 if i < executed else 0, + "units": [{"unit": "unsupported", "status": "unsupported"}], + } + if i < executed: + record["units"].append( + {"unit": "executed", "status": "path", "instrument": {"bytes": 20, "jvm": [{"seconds": 8, "success": True}]}} + ) + (baseline / f"{i}.json").write_text(json.dumps(record)) + (current / f"{i}.json").write_text(json.dumps(record)) + + if executed == 3: + with pytest.raises(ValueError, match="engine proof requires exactly two pins"): + p.engine_proof(current, baseline) + else: + proof = p.engine_proof(current, baseline) + assert len(proof["pins"]) == 2 + assert proof["differences"] == 0