From c6e87a851764bc1a9bc64572775e053580bae2d7 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sun, 4 Oct 2026 20:19:14 +0200 Subject: [PATCH] unseen-eval groups 4-6: generic AX batch dispatcher (engine_trace on it), replay task and image stage, scorer with adoption gate Implemented with Codex. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/engine-image.yml | 6 + .kiro/specs/unseen-repo-evaluation/tasks.md | 50 +- benchmarks/ax/batch.py | 434 +++++++++ benchmarks/learn/engine_trace.py | 76 +- benchmarks/learn/engine_trace_ax.py | 265 +----- .../2026-10-04-unseen-06-scorer/record.json | 41 + benchmarks/unseen/README.md | 92 ++ benchmarks/unseen/fixtures/scorer.json | 860 ++++++++++++++++++ benchmarks/unseen/proof.py | 128 +++ benchmarks/unseen/replay.py | 193 ++++ benchmarks/unseen/replay_entry.py | 219 +++++ benchmarks/unseen/score.py | 308 +++++++ plane/Dockerfile.engine-task | 10 +- src/openultrasast/plane/memory.py | 2 + tests/test_ax_batch.py | 159 ++++ tests/test_unseen_image.py | 16 + tests/test_unseen_proof.py | 56 ++ tests/test_unseen_replay.py | 79 ++ tests/test_unseen_replay_entry.py | 136 +++ tests/test_unseen_score.py | 127 +++ 20 files changed, 2952 insertions(+), 305 deletions(-) create mode 100644 benchmarks/ax/batch.py create mode 100644 benchmarks/measurements/2026-10-04-unseen-06-scorer/record.json create mode 100644 benchmarks/unseen/fixtures/scorer.json create mode 100644 benchmarks/unseen/proof.py create mode 100644 benchmarks/unseen/replay.py create mode 100644 benchmarks/unseen/replay_entry.py create mode 100644 benchmarks/unseen/score.py create mode 100644 tests/test_ax_batch.py create mode 100644 tests/test_unseen_image.py create mode 100644 tests/test_unseen_proof.py create mode 100644 tests/test_unseen_replay.py create mode 100644 tests/test_unseen_replay_entry.py create mode 100644 tests/test_unseen_score.py diff --git a/.github/workflows/engine-image.yml b/.github/workflows/engine-image.yml index dca961c..764f092 100644 --- a/.github/workflows/engine-image.yml +++ b/.github/workflows/engine-image.yml @@ -7,6 +7,7 @@ on: - 'src/**' - 'benchmarks/learn/engine_trace*.py' - 'benchmarks/learn/engine_task_entry.py' + - 'benchmarks/unseen/replay_entry.py' - 'benchmarks/learn/engine_worker_service.py' - 'Dockerfile' - 'plane/Dockerfile.engine-task' @@ -30,6 +31,10 @@ jobs: dockerfile: Dockerfile - package: ousast-engine-task dockerfile: plane/Dockerfile.engine-task + target: engine + - package: ousast-replay-task + dockerfile: plane/Dockerfile.engine-task + target: replay steps: - uses: actions/checkout@v5 - uses: docker/setup-buildx-action@v3 @@ -50,6 +55,7 @@ jobs: with: context: . file: ${{ matrix.dockerfile }} + target: ${{ matrix.target }} platforms: linux/amd64 push: true tags: | diff --git a/.kiro/specs/unseen-repo-evaluation/tasks.md b/.kiro/specs/unseen-repo-evaluation/tasks.md index 9195df9..d18b477 100644 --- a/.kiro/specs/unseen-repo-evaluation/tasks.md +++ b/.kiro/specs/unseen-repo-evaluation/tasks.md @@ -158,7 +158,7 @@ Execution rules: - _Requirements: 1.1, 1.2, 1.3_ - [ ] 4. Dispatcher (design 5.3; increment 4) -- [ ] 4.1 Items, workloads and the AX lane +- [x] 4.1 Items, workloads and the AX lane - `benchmarks/ax/batch.py`: `Item` (id, input objects, result object name, command, extra env, input digest), `Workload` (image digest, allowed URL env names, deadline, result validator) and the `ax` lane, moved from `benchmarks/learn/engine_trace_ax.py`: manifest render, egress readiness, presigned GET and PUT as `.dat`, @@ -167,7 +167,7 @@ Execution rules: - `tests/test_ax_batch.py` with a fake `ax` CLI and a fake store: manifest holds only the allowed URL env names and no credential; names carry the `ousast-engine-` prefix; cleanup on success and failure. - _Requirements: 4.1, 4.2, 4.5_ -- [ ] 4.2 Scheduler, fallback and resume +- [x] 4.2 Scheduler, fallback and resume - `batch.py`: two AX lanes and one VM `docker` lane by default (one container at a time); sandbox death retried once under a new name, a second death or a JVM OOM goes to the VM fallback queue ahead of overflow; `progress.json` written by the scheduler only; a `STOP` file halts launches; resume skips items whose stored @@ -175,7 +175,7 @@ Execution rules: - Tests with fake lanes: retry then fallback; OOM straight to fallback; resume skips; `STOP`; cleanup failure halts; never more than two AX tasks in flight. - _Requirements: 4.3_ -- [ ] 4.3 Engine pass on the dispatcher +- [x] 4.3 Engine pass on the dispatcher - `benchmarks/learn/engine_trace_ax.py` and `engine_trace.run_plan` become an engine `Workload` over `batch.py`; no behaviour change. - Observable: `tests/test_ax_executor.py` and `tests/test_engine_trace.py` pass with their assertions unchanged @@ -188,14 +188,14 @@ Execution rules: - _Requirements: 4.3_ - [ ] 5. Replay task (design 5.1, 5.2; increment 5) -- [ ] 5.1 Image stage +- [x] 5.1 Image stage - `plane/Dockerfile.engine-task`: second stage `FROM engine AS replay` adding `git`, the package with its runtime dependencies and `benchmarks/unseen/replay_entry.py`; smoke `ousast pre-push --help`. The root image stays `ghcr.io/norandom/ax-task-runner:v0.3.1`; the engine stage is unchanged. - `tests/test_unseen_image.py`: parses the Dockerfile; the `replay` stage builds from `engine`; the root image pin; the engine stage's text is byte-identical to before. - _Requirements: 4.1_ -- [ ] 5.2 Task entry +- [x] 5.2 Task entry - `benchmarks/unseen/replay_entry.py`, standard library only, reusing `engine_task_entry.transfer`: GET the spec; `git init` and depth-1 fetches of head and base over HTTPS (180 s bound), verified with `git cat-file -e`; run `ousast pre-push` with the spec's hook flags (defaults, or `--deadline 300` for the long-deadline arm), heap @@ -206,7 +206,7 @@ Execution rules: fields; a missing commit, a failed fetch and a hook crash each give a failure record; zero bytes read and a replay faster than JVM start are flagged `suspect_fast`. - _Requirements: 3.1, 4.1, 4.2_ -- [ ] 5.3 Replay workload +- [x] 5.3 Replay workload - `benchmarks/unseen/replay.py`: builds `Item`s from a frozen slice and an arm (`default` or `long_deadline`); the spec object carries the repository URL so names stay out of AX status and telemetry; task names `ousast-engine-replay--` with an opaque slug; the validator rejects a record without @@ -223,28 +223,28 @@ Execution rules: exit codes, the quick-tier comparison. - _Requirements: 4.1, 4.2, 4.5_ -- [ ] 6. Scorer (design 3, 4, 7.1, 7.3; increment 6) -- [ ] 6.1 Coverage states +- [x] 6. Scorer (design 3, 4, 7.1, 7.3; increment 6) +- [x] 6.1 Coverage states - `benchmarks/unseen/score.py`: each record gets one state (`instrument_failure`, `unanalysable`, `quick_only`, `engine_covered`) from `instrument.json` and the artifact; `suspect_fast` goes to failure; a slice above 5% `instrument_failure` after re-runs is refused. - `tests/test_unseen_score.py` on synthetic records: every state; the refusal; a failure never counts as quiet or missed. - _Requirements: 3.3_ -- [ ] 6.2 Catch and false-alarm rules +- [x] 6.2 Catch and false-alarm rules - `score.py`: family match (engine family, quick-rule CWE through `taxonomy.py`), the location table of design 4.1, all three streams with the stream named per catch; false alarm on any finding of any stream; the alerts-only rate. - Tests: right family wrong place misses; `access_control` on the handler; `config_secrets` file scope; a quick-rule-only catch; one advisory finding on an ordinary change is a false alarm. - _Requirements: 3.2_ -- [ ] 6.3 Rates, intervals and breakdowns +- [x] 6.3 Rates, intervals and breakdowns - `score.py`: Wilson 95% intervals; repository-cluster bootstrap for false alarms; ICC; breakdowns per family, method, `post_cutoff`, stream, coverage state and the confirmed subset. - Tests: the n values of design section 3 (315, 483, 756, 62, 88, 97) reproduced; bootstrap seeded; ICC on a synthetic clustered set. - _Requirements: 2.3, 3.3_ -- [ ] 6.4 Budgets, paired tests, the gate and record 06 +- [x] 6.4 Budgets, paired tests, the gate and record 06 - `score.py`: per-budget thresholds (1%, 5%, 10%), labelled `exploratory` in-slice and `fixed` when supplied; a binary arm's single operating point with the budgets it meets; exact two-sided McNemar for catch and for false alarm; `gate(A, B)` returning `pass | regression | not_run` per design 7.3. @@ -354,3 +354,31 @@ Validation: requested pytest selection 103 passed, 1 skipped, exit 0 (population output discarded); `ruff check .` exit 0; `ruff format --check .` exit 0 (411 files); `mypy src/openultrasast` exit 0 (160 source files). The CLI `--help` smoke exited 0. Source-read instrumentation printed 20,101 bytes for the extraction implementation. + +### Groups 4–6 offline implementation (2026-10-04) + +Tasks 4.1–4.3, 5.1–5.3 and 6.1–6.4 are implemented and independently reviewed. +Tasks 4.4 and 5.4 remain coordinator-owned network proofs; group headers 4 and 5 stay +pending those exits. No commit, network request, Docker, AX or kubectl command was run. +The sourcing job's draft.py, source.py, extract.py and journal.py are unchanged. + +The shared dispatcher retains the engine adapter's historical questions.json object +name to preserve its unchanged assertions; replay uses spec.dat and result.dat. +The engine Dockerfile body is unchanged except its required AS engine alias; CI builds +engine and replay separately. S3Store's read_only class default preserves existing +constructor-bypassing transport fixtures; initialized read-only stores still refuse writes. + +Exact coordinator commands and input contracts are in benchmarks/unseen/README.md, +including frozen TOML/private pointer verification, scoring joins, own-history proof +preparation and anonymous measurement-record generation. JVM wall time is the hook's +aggregate scan build/query timing and is labelled as such. Unsupported-only partitions +with explicit language_not_covered evidence do not pretend to have launched a JVM. + +Record 06 contains the committed synthetic fixture and scorer-output digests. The +reserved-repository guard was run as a black box with both output streams discarded. +Independent review: APPROVED; all reported integration and coverage defects resolved. + +Final verification: requested pytest command 279 passed, 8 skipped, exit 0; separate +`tests/test_ax_batch.py` 5 passed, exit 0; `ruff check .`, `ruff format --check .`, +`mypy src/openultrasast`, and the output-discarded reserved-repository guard all exit 0. +The own-history proof preparer and replay dry run read five items and exit 0. diff --git a/benchmarks/ax/batch.py b/benchmarks/ax/batch.py new file mode 100644 index 0000000..039b584 --- /dev/null +++ b/benchmarks/ax/batch.py @@ -0,0 +1,434 @@ +"""Host-only AX Task transport; credentials never enter task manifests.""" + +from __future__ import annotations + +import hashlib +import ipaddress +import json +import os +import queue +import re +import shutil +import subprocess +import threading +import time +import uuid +from collections import Counter, deque +from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait +from dataclasses import dataclass, field +from datetime import timedelta +from pathlib import Path +from urllib.parse import urlsplit + +import yaml + +from openultrasast.config import load_dotenv +from openultrasast.plane.memory import PRESIGNED_ARCHIVE_SUFFIX, S3Store, open_store + + +class EgressError(ValueError): + """Safe diagnostic without presigned URL contents.""" + + +class SandboxFailure(Exception): + """A terminal AX task without an authoritative result object.""" + + +def public_endpoint(url): + """Reject endpoints the actor's hostname-based HTTPS egress policy cannot reach.""" + parts = urlsplit(url) + host = parts.hostname or "" + try: + ipaddress.ip_address(host) + is_ip = True + except ValueError: + is_ip = False + if ( + parts.scheme != "https" + or not host + or is_ip + or "." not in host + or host.endswith((".localhost", ".local", ".internal")) + or parts.port not in (None, 443) + or parts.username + or parts.password + ): + raise EgressError( + "AX requires a public HTTPS DNS hostname in S3_ENDPOINT and presigned URLs: " + "the egress policy allows the store hostname on port 443 only (no IP or plain HTTP)" + ) + return host.lower() + + +def task_state(text, name): + """Accept AX YAML/JSON output and its human-readable tasks table.""" + try: + doc = yaml.safe_load(text) + except yaml.YAMLError: + doc = None + rows = doc.get("items", doc.get("tasks", [doc])) if isinstance(doc, dict) else doc if isinstance(doc, list) else [] + for row in rows: + if not isinstance(row, dict) or row.get("metadata", {}).get("name", row.get("name")) != name: + continue + status = row.get("status", {}) + ip = next( + (status.get(key) or row.get(key) for key in ("workerIP", "workerIp", "worker_ip", "ip") if status.get(key) or row.get(key)), + None, + ) + worker = status.get("worker") + if not ip and isinstance(worker, dict): + ip = worker.get("ip") or worker.get("address") + return str(status.get("phase", row.get("phase", ""))), ip + lines = [line.split() for line in text.splitlines() if line.strip()] + if lines: + headers = [re.sub(r"[^a-z]", "", cell.lower()) for cell in lines[0]] + for cells in lines[1:]: + if name not in cells: + continue + row = dict(zip(headers, cells, strict=False)) + phase = row.get("phase", row.get("status", "")) + ip = row.get("workerip") or row.get("ip") + if not ip: + for cell in cells: + try: + ipaddress.ip_address(cell) + ip = cell + break + except ValueError: + pass + return phase, None if ip in {"", "-", ""} else ip + listed = isinstance(doc, (dict, list)) or bool(lines and "name" in headers and {"phase", "status"}.intersection(headers)) + return "Missing" if listed else "", None + + +def bounded_read(read, timeout): + """Do not let SDK retries postpone task teardown beyond the pin deadline. + + Only the read runs in a daemon thread; it cannot publish a late result to disk. + No executor shutdown waits for an SDK call whose socket has stopped responding. + """ + if timeout <= 0: + raise TimeoutError("AX result deadline") + replies = queue.Queue(maxsize=1) + + def fetch(): + try: + replies.put((True, read())) + except Exception as exc: + replies.put((False, exc)) + + threading.Thread(target=fetch, daemon=True, name="engine-ax-result-read").start() + try: + ok, result = replies.get(timeout=timeout) + except queue.Empty: + raise TimeoutError("AX result deadline") from None + if not ok: + raise result + return result + + +class AXLane: + def __init__(self, args, workload, store=None, pause=time.sleep, clock=time.monotonic): + self.workload = workload + self.args = args + self.pause, self.clock = pause, clock + self.run = uuid.uuid4().hex + load_dotenv() + self.host = public_endpoint(os.environ.get("S3_ENDPOINT", "")) + # Validate the image before opening a store, whose constructor probes the bucket. + validate_image(workload.image) + self.store = store if store is not None else open_store() + if store is None and not isinstance(self.store, S3Store): + raise ValueError("AX requires OUSAST_MEMORY=s3://bucket[/prefix]") + + def __call__(self, item, output, attempt): + attempts = [] + try: + done, worker = self._execute_attempt(item, output, attempt, attempts) + except SandboxFailure: + return {"status": "sandbox_failure", "ax_attempts": attempts} + if done is None or done.returncode: + return {"status": "instrument_failure", "ax_attempts": attempts} + record = json.loads((output / "result.json").read_text()) + return {**record, "worker_ip": worker, "ax_attempts": attempts} + + def ax(self, *args, manifest=None, timeout=30): + env = dict(os.environ) + if self.args.kubeconfig: + env["KUBECONFIG"] = str(Path(self.args.kubeconfig).expanduser()) + done = subprocess.run( + [str(Path(self.args.ax_bin).expanduser()), *args, *(["--atespace", self.args.atespace] if args[0] != "apply" else [])], + input=json.dumps(manifest) if manifest is not None else None, + env=env, + capture_output=True, + text=True, + timeout=max(0.01, timeout), + ) + if done.returncode: + # AX errors may echo manifest URLs; report only verb and exit code. + raise ValueError(f"ax {args[0]} failed (exit {done.returncode})") + return done + + def _execute_attempt(self, item, output, attempt, attempts): + pin_id = item.id + digest = hashlib.sha256(f"{self.run}/{pin_id}/attempt-{attempt}".encode()).hexdigest()[:12] + slug = re.sub(r"[^a-z0-9]+", "-", str(pin_id).lower()).strip("-")[:22] + name = f"ousast-engine-{slug or 'pin'}-{digest}" + key_id = hashlib.sha256(str(pin_id).encode()).hexdigest() + prefix = f"engine-queue/{self.run}/{key_id}/attempt-{attempt}/" + keys = [prefix + item.object_names.get(key, key.lower() + PRESIGNED_ARCHIVE_SUFFIX) for key in item.inputs] + keys.append(prefix + item.result_object) + expires = timedelta(seconds=self.workload.deadline + 600) + worker_ip = None + submitted = False + started = self.clock() + phase = "" + try: + for key, data in zip(keys, item.inputs.values(), strict=False): + self.store._put(key, data) + urls = {env: self.store.presign_get(key, expires) for env, key in zip(item.inputs, keys, strict=False)} + urls["RESULT_URL"] = self.store.presign_put(keys[-1], expires) + if any(public_endpoint(url) != self.host for url in urls.values()): + raise EgressError("AX presign hostname must match S3_ENDPOINT for the store egress policy") + document = manifest(name, self.workload, item, urls, self.args.atespace) + end = self.clock() + self.workload.deadline + # An apply timeout can still have created the task; cleanup must cover it. + submitted = True + self.ax("apply", "-f", "-", manifest=document, timeout=min(30, self.workload.deadline)) + while self.clock() < end: + result = bounded_read(lambda: self.store._get(keys[-1]), end - self.clock()) + if self.clock() >= end: + return None, worker_ip + phase = "" + try: + state = self.ax("get", "tasks", timeout=min(30, end - self.clock())).stdout + phase, ip = task_state(state, name) + worker_ip = ip or worker_ip + except (ValueError, OSError, subprocess.SubprocessError): + # The object is authoritative even when AX status is temporarily unavailable. + pass + if self.clock() >= end: + return None, worker_ip + if result is not None: + self.workload.validate(result[0], output) + return subprocess.CompletedProcess([], 0, "", ""), worker_ip + if phase.lower() in {"failed", "missing"}: + # The worker may have uploaded while we were reading AX status. + result = bounded_read(lambda: self.store._get(keys[-1]), end - self.clock()) + if self.clock() >= end: + return None, worker_ip + if result is not None: + self.workload.validate(result[0], output) + return subprocess.CompletedProcess([], 0, "", ""), worker_ip + raise SandboxFailure + self.pause(min(2, max(0, end - self.clock()))) + return None, worker_ip + except (subprocess.TimeoutExpired, TimeoutError): + return None, worker_ip + except SandboxFailure: + raise + except Exception as exc: + # Never expose exception strings which may contain object-scoped bearer URLs. + detail = str(exc) if isinstance(exc, EgressError) else type(exc).__name__ + return subprocess.CompletedProcess([], 1, "", "AX transport failed: " + detail), worker_ip + finally: + attempts.append({"name": name, "phase": phase, "seconds": self.clock() - started}) + errors = [] + if submitted: + try: + self.ax("delete", "task", name) + except Exception as exc: + # A successful list already confirmed a disappeared task is gone. + if phase != "Missing": + errors.append("task: " + type(exc).__name__) + for key in keys: + try: + self.store._delete(key) + except Exception as exc: + errors.append("object: " + type(exc).__name__) + if errors: + raise RuntimeError("AX cleanup failed; stop dispatching: " + ", ".join(errors)) + + +@dataclass +class Item: + id: str + inputs: dict[str, bytes] + result_object: str + command: tuple[str, ...] + extra_env: dict[str, str] + input_digest: str + object_names: dict[str, str] = field(default_factory=dict) + + +@dataclass(frozen=True) +class Workload: + image: str + url_env: frozenset[str] + deadline: float + validate: object + + +def validate_image(image): + if not re.fullmatch(r"[^\s@]+@sha256:[0-9a-f]{64}", image): + raise ValueError("AX --image must be pinned by sha256 digest") + + +def manifest(name, workload, item, urls, atespace="default", *, validate_name=True): + validate_image(workload.image) + if validate_name and (not re.fullmatch(r"ousast-engine-[a-z0-9-]+", name) or len(name.encode()) > 49): + raise ValueError("invalid AX task name") + if set(urls) != workload.url_env: + raise ValueError("unexpected URL environment names") + # Extra env is deliberately a small numeric execution contract, never inherited credentials. + if set(item.extra_env) - {"DEADLINE", "QUESTION_DEADLINE"} or any( + not re.fullmatch(r"[0-9]+(?:\.[0-9]+)?", str(v)) for v in item.extra_env.values() + ): + raise ValueError("unexpected task environment") + for url in urls.values(): + public_endpoint(url) + env = [{"name": key, "value": str(value)} for key, value in {**urls, **item.extra_env}.items()] + if len(json.dumps(env).encode()) >= 32768: + raise ValueError("AX Task env exceeds the 32 KB limit") + return { + "apiVersion": "ax.io/v1alpha1", + "kind": "Task", + "metadata": {"name": name, "atespace": atespace}, + "spec": {"image": workload.image, "env": env, "command": list(item.command)}, + } + + +def write_json(path, record): + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(".tmp") + temporary.write_text(json.dumps(record, sort_keys=True) + "\n") + temporary.replace(path) + + +def schedule(pending, lanes, out, execute, complete, *, fallbacks=(), overflow=True, progress=None): + """The sole progress writer; callbacks carry workload-specific records, never transport.""" + if sum(lane in {"ax", "kind"} for lane in lanes) > 2 or lanes.count("docker") > 1: + raise ValueError("at most two AX lanes and one VM lane") + pending, fallbacks = deque(pending), deque(fallbacks) + progress = progress if progress is not None else {"pins_done": 0, "pins_total": len(pending) + len(fallbacks)} + progress["lanes"] = dict(Counter(lanes)) + active = {} + failures = False + + def persist(): + progress.update(pins_in_flight=len(active), active_lanes=dict(Counter(lane for _, lane in active.values()))) + if any(lane in {"ax", "kind"} for lane in lanes): + progress["fallbacks_pending"] = len(fallbacks) + write_json(out / "progress.json", progress) + + with ThreadPoolExecutor(max_workers=len(lanes)) as pool: + + def launch(lane): + if (out / "STOP").exists(): + return + if lane == "docker" and fallbacks: + item, previous = fallbacks.popleft() + elif pending and (lane != "docker" or overflow): + item, previous = pending.popleft(), None + else: + return + active[pool.submit(execute, item, lane, previous)] = item, lane + + for lane in lanes: + launch(lane) + persist() + try: + while active: + finished, _ = wait(active, return_when=FIRST_COMPLETED) + # Resolve ALL completed futures before launching; cleanup exceptions stop launches. + results = [(active.pop(future), future.result()) for future in finished] + for (item, lane), result in results: + failed, fallback = complete(item, lane, result, progress) + if fallback is not None: + fallbacks.append((item, fallback)) + else: + failures |= failed + progress["pins_done"] += 1 + for lane, count in (Counter(lanes) - Counter(lane for _, lane in active.values())).items(): + for _ in range(count): + launch(lane) + persist() + finally: + persist() + return failures + + +def dispatch(items, workload, out, executors, *, lanes=("ax", "ax", "docker"), overflow=True): + """Digest checkpoints, one instrument rerun, sandbox retry, priority VM fallback.""" + out.mkdir(parents=True, exist_ok=True) + pending, fallbacks = [], [] + + def path(item): + return out / (hashlib.sha256(item.id.encode()).hexdigest() + ".json") + + for item in items: + previous = json.loads(path(item).read_text()) if path(item).exists() else {} + if previous.get("input_digest") == item.input_digest: + if previous.get("fallback_pending"): + fallbacks.append((item, previous)) + continue + if previous.get("done"): + continue + pending.append(item) + + def execute(item, lane, previous): + output = out / hashlib.sha256(item.id.encode()).hexdigest() + output.mkdir(exist_ok=True) + attempts = [] + for attempt in range(2): + attempt_output = output / ("attempt-" + uuid.uuid4().hex) + attempt_output.mkdir() + record = executors[lane](item, attempt_output, attempt) + 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", [])}) + 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} + + def complete(item, lane, record, progress): + fallback = lane in {"ax", "kind"} and record.get("status") in {"sandbox_failure", "oom"} and "docker" in lanes + record["fallback_pending"] = fallback + write_json(path(item), record) + progress["last_item"] = hashlib.sha256(item.id.encode()).hexdigest() + return record.get("status") not in {"ok", "unanalysable", "quick_only", "engine_covered"}, record if fallback else None + + return schedule(pending, lanes, out, execute, complete, fallbacks=fallbacks, overflow=overflow) + + +class DockerLane(AXLane): + """VM lane uses the same presigned-object protocol and pinned replay image.""" + + def _execute_attempt(self, item, output, attempt, attempts): + # Reuse all object/validation/cleanup logic, replacing only the task runner. + import copy + + lane = copy.copy(self) + lane.ax = lane._docker_command + lane._docker_name = None + return AXLane._execute_attempt(lane, item, output, attempt, attempts) + + 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]] + 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: + 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") + if args[0] == "get": + phase = "Running" if done.stdout.startswith("true") else "Failed" + 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 15d56a0..ebb3eaa 100644 --- a/benchmarks/learn/engine_trace.py +++ b/benchmarks/learn/engine_trace.py @@ -16,13 +16,13 @@ import os import re import shutil import subprocess +import sys import tempfile import textwrap import threading import time import tomllib -from collections import Counter, defaultdict, deque -from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait +from collections import Counter, defaultdict from pathlib import Path from engine_trace_k8s import IMAGE, Kubernetes @@ -39,6 +39,9 @@ from openultrasast.plane.memory import FileStore from openultrasast.plane.tasks.alerts import engine_languages from openultrasast.preprocess import detect_language +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) +from benchmarks.ax.batch import schedule + ROOT = Path(__file__).resolve().parents[2] DEFAULT_OUT = Path.home() / "ousast-results/plane/engine-trace" SNAPSHOT_FILES = ( @@ -47,6 +50,7 @@ SNAPSHOT_FILES = ( "benchmarks/learn/engine_trace_k8s.py", "benchmarks/learn/engine_trace_queue.py", "benchmarks/learn/engine_trace_ax.py", + "benchmarks/ax/batch.py", "benchmarks/learn/engine_task_entry.py", "benchmarks/learn/engine_worker_service.py", "benchmarks/learn/engine_trace_parse.py", @@ -791,8 +795,8 @@ def run_plan(plan, source, provenance, inputs, args): "active_lanes": {}, } counts = Counter() - pending = deque(plan) - fallbacks = deque() + pending = list(plan) + fallbacks = [] if args.executor in {"ax", "mixed-ax"}: pending.clear() for pin in plan: @@ -803,12 +807,6 @@ def run_plan(plan, source, provenance, inputs, args): fallbacks.append((pin, old)) else: pending.append(pin) - active = {} - failures = False - - def persist(): - progress.update(pins_in_flight=len(active), active_lanes=dict(Counter(lane for _, lane in active.values()))) - write_json(args.out / "progress.json", progress) def execute(pin, lane, attempt=None): if not any(u["supported"] for u in pin["units"]): @@ -825,52 +823,20 @@ def run_plan(plan, source, provenance, inputs, args): options.image = getattr(args, "docker_image", args.image) return run_pin(pin, source, provenance, inputs, args.out, options) - with ThreadPoolExecutor(max_workers=len(lanes)) as pool: + def complete(pin, lane, failed, progress): + result = json.loads((args.out / pin_name(pin)).read_text()) + selected = {u["unit"] for u in pin["units"]} + reasons = [u.get("reason", "") for u in result["units"] if u["unit"] in selected] + sandbox_failed = args.executor == "mixed-ax" and result.get("container_exit") == 1 and "AX sandbox failed twice" in reasons + if lane == "ax" and (sandbox_failed or any("JVM out of memory" in reason for reason in reasons)): + result.update(fallback="vm-sandbox" if sandbox_failed else "vm-oom", fallback_pending=True) + write_json(args.out / pin_name(pin), result) + return False, result + counts.update(u["status"] for u in result["units"] if u["unit"] in selected) + progress.update(last_pin={"repo": pin["repo"], "pin": pin["pin"]}, statuses=dict(counts)) + return failed, None - def launch(lane): - if (args.out / "STOP").exists(): - return - attempt = None - if lane == "docker" and fallbacks: - pin, attempt = fallbacks.popleft() - elif pending and not (args.executor == "ax" and lane == "docker"): - pin = pending.popleft() - else: - return - active[pool.submit(execute, pin, lane, attempt)] = (pin, lane) - - for lane in lanes: - launch(lane) - persist() - while active: - completed, _ = wait(active, return_when=FIRST_COMPLETED) - for future in completed: - pin, lane = active.pop(future) - failed = future.result() - result = json.loads((args.out / pin_name(pin)).read_text()) - selected = {u["unit"] for u in pin["units"]} - reasons = [u.get("reason", "") for u in result["units"] if u["unit"] in selected] - sandbox_failed = args.executor == "mixed-ax" and result.get("container_exit") == 1 and "AX sandbox failed twice" in reasons - if lane == "ax" and (sandbox_failed or any("JVM out of memory" in reason for reason in reasons)): - result.update(fallback="vm-sandbox" if sandbox_failed else "vm-oom", fallback_pending=True) - write_json(args.out / pin_name(pin), result) - fallbacks.append((pin, result)) - continue - failures |= failed - selected = {u["unit"] for u in pin["units"]} - counts.update(u["status"] for u in result["units"] if u["unit"] in selected) - progress.update( - pins_done=progress["pins_done"] + 1, last_pin={"repo": pin["repo"], "pin": pin["pin"]}, statuses=dict(counts) - ) - idle = Counter(lanes) - Counter(lane for _, lane in active.values()) - # Give a waiting fallback priority over overflow on the single VM lane. - for lane, count in idle.items(): - for _ in range(count): - launch(lane) - if args.executor in {"ax", "mixed-ax"}: - progress["fallbacks_pending"] = len(fallbacks) - persist() - return failures + return schedule(pending, lanes, args.out, execute, complete, fallbacks=fallbacks, overflow=args.executor != "ax", progress=progress) def main(): diff --git a/benchmarks/learn/engine_trace_ax.py b/benchmarks/learn/engine_trace_ax.py index d6317ce..6046f8b 100644 --- a/benchmarks/learn/engine_trace_ax.py +++ b/benchmarks/learn/engine_trace_ax.py @@ -1,184 +1,50 @@ -"""Host-only AX Task transport; credentials never enter task manifests.""" +"""Engine workload adapter for the shared AX transport.""" from __future__ import annotations -import hashlib import io -import ipaddress import json -import os -import queue -import re import shutil import subprocess +import sys import tarfile import tempfile -import threading import time -import uuid -from datetime import timedelta from pathlib import Path -from urllib.parse import urlsplit -import yaml +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from engine_trace_k8s import pack -from openultrasast.config import load_dotenv -from openultrasast.plane.memory import PRESIGNED_ARCHIVE_SUFFIX, S3Store, open_store - - -class EgressError(ValueError): - """Safe diagnostic without presigned URL contents.""" - - -class SandboxFailure(Exception): - """A terminal AX task without an authoritative result object.""" - - -def public_endpoint(url): - """Reject endpoints the actor's hostname-based HTTPS egress policy cannot reach.""" - parts = urlsplit(url) - host = parts.hostname or "" - try: - ipaddress.ip_address(host) - is_ip = True - except ValueError: - is_ip = False - if ( - parts.scheme != "https" - or not host - or is_ip - or "." not in host - or host.endswith((".localhost", ".local", ".internal")) - or parts.port not in (None, 443) - or parts.username - or parts.password - ): - raise EgressError( - "AX requires a public HTTPS DNS hostname in S3_ENDPOINT and presigned URLs: " - "the egress policy allows the store hostname on port 443 only (no IP or plain HTTP)" - ) - return host.lower() +from benchmarks.ax.batch import AXLane, Item, SandboxFailure, Workload, manifest +from benchmarks.ax.batch import task_state as task_state +from openultrasast.plane.memory import open_store as open_store def task_manifest(name, args, urls): - if not re.fullmatch(r"[^\s@]+@sha256:[0-9a-f]{64}", args.image): - raise ValueError("AX --image must be pinned by sha256 digest") - if set(urls) != {"SOURCE_URL", "QUESTIONS_URL", "RESULT_URL"}: - raise ValueError("AX Task accepts only source, questions and result presigned URLs") - for url in urls.values(): - public_endpoint(url) - env = [{"name": key, "value": value} for key, value in urls.items()] - env.extend( - {"name": key, "value": str(value)} - for key, value in {"DEADLINE": args.deadline, "QUESTION_DEADLINE": args.question_deadline}.items() + return manifest( + name, engine_workload(args, lambda *_: None), engine_item({}, "pin", b"", args), urls, args.atespace, validate_name=False ) - if len(json.dumps(env).encode()) >= 32768: - raise ValueError("AX Task env exceeds the 32 KB limit") - return { - "apiVersion": "ax.io/v1alpha1", - "kind": "Task", - "metadata": {"name": name, "atespace": args.atespace}, - "spec": {"image": args.image, "env": env, "command": ["python3", "/app/benchmarks/learn/engine_task_entry.py"]}, - } -def task_state(text, name): - """Accept AX YAML/JSON output and its human-readable tasks table.""" - try: - doc = yaml.safe_load(text) - except yaml.YAMLError: - doc = None - rows = doc.get("items", doc.get("tasks", [doc])) if isinstance(doc, dict) else doc if isinstance(doc, list) else [] - for row in rows: - if not isinstance(row, dict) or row.get("metadata", {}).get("name", row.get("name")) != name: - continue - status = row.get("status", {}) - ip = next( - (status.get(key) or row.get(key) for key in ("workerIP", "workerIp", "worker_ip", "ip") if status.get(key) or row.get(key)), - None, - ) - worker = status.get("worker") - if not ip and isinstance(worker, dict): - ip = worker.get("ip") or worker.get("address") - return str(status.get("phase", row.get("phase", ""))), ip - lines = [line.split() for line in text.splitlines() if line.strip()] - if lines: - headers = [re.sub(r"[^a-z]", "", cell.lower()) for cell in lines[0]] - for cells in lines[1:]: - if name not in cells: - continue - row = dict(zip(headers, cells, strict=False)) - phase = row.get("phase", row.get("status", "")) - ip = row.get("workerip") or row.get("ip") - if not ip: - for cell in cells: - try: - ipaddress.ip_address(cell) - ip = cell - break - except ValueError: - pass - return phase, None if ip in {"", "-", ""} else ip - listed = isinstance(doc, (dict, list)) or bool(lines and "name" in headers and {"phase", "status"}.intersection(headers)) - return "Missing" if listed else "", None +def engine_workload(args, validate): + return Workload(args.image, frozenset({"SOURCE_URL", "QUESTIONS_URL", "RESULT_URL"}), args.deadline, validate) -def bounded_read(read, timeout): - """Do not let SDK retries postpone task teardown beyond the pin deadline. - - Only the read runs in a daemon thread; it cannot publish a late result to disk. - No executor shutdown waits for an SDK call whose socket has stopped responding. - """ - if timeout <= 0: - raise TimeoutError("AX result deadline") - replies = queue.Queue(maxsize=1) - - def fetch(): - try: - replies.put((True, read())) - except Exception as exc: - replies.put((False, exc)) - - threading.Thread(target=fetch, daemon=True, name="engine-ax-result-read").start() - try: - ok, result = replies.get(timeout=timeout) - except queue.Empty: - raise TimeoutError("AX result deadline") from None - if not ok: - raise result - return result +def engine_item(pin, pin_id, source, args): + return Item( + pin_id, + {"SOURCE_URL": source, "QUESTIONS_URL": json.dumps(pin).encode()}, + "result.dat", + ("python3", "/app/benchmarks/learn/engine_task_entry.py"), + {"DEADLINE": str(args.deadline), "QUESTION_DEADLINE": str(args.question_deadline)}, + "", + {"SOURCE_URL": "source.dat", "QUESTIONS_URL": "questions.json"}, + ) -class AX: +class AX(AXLane): def __init__(self, args, store=None, pause=time.sleep, clock=time.monotonic): - self.args = args - self.pause, self.clock = pause, clock - self.run = uuid.uuid4().hex - load_dotenv() - self.host = public_endpoint(os.environ.get("S3_ENDPOINT", "")) - # Validate the image before opening a store, whose constructor probes the bucket. - task_manifest("validate", args, {key: f"https://{self.host}/" for key in ("SOURCE_URL", "QUESTIONS_URL", "RESULT_URL")}) - self.store = store if store is not None else open_store() - if store is None and not isinstance(self.store, S3Store): - raise ValueError("AX requires OUSAST_MEMORY=s3://bucket[/prefix]") - - def ax(self, *args, manifest=None, timeout=30): - env = dict(os.environ) - if self.args.kubeconfig: - env["KUBECONFIG"] = str(Path(self.args.kubeconfig).expanduser()) - done = subprocess.run( - [str(Path(self.args.ax_bin).expanduser()), *args, *(["--atespace", self.args.atespace] if args[0] != "apply" else [])], - input=json.dumps(manifest) if manifest is not None else None, - env=env, - capture_output=True, - text=True, - timeout=max(0.01, timeout), - ) - if done.returncode: - # AX errors may echo manifest URLs; report only verb and exit code. - raise ValueError(f"ax {args[0]} failed (exit {done.returncode})") - return done + super().__init__(args, engine_workload(args, lambda *_: None), store, pause, clock) def receive(self, data, output, pin): # Validate in isolation: malformed results must never become usable checkpoints. @@ -220,11 +86,15 @@ class AX: shutil.copytree(root, output, dirs_exist_ok=True) def execute(self, checkout, output, pin, pin_id): + import copy + + self = copy.copy(self) + self.workload = engine_workload(self.args, lambda data, target: self.receive(data, target, pin)) attempts = [] try: for attempt in range(2): try: - return self._execute_attempt(checkout, output, pin, pin_id, attempt, attempts) + return self._execute_attempt(engine_item(pin, pin_id, pack(checkout), self.args), output, attempt, attempts) except SandboxFailure: if attempt == 0: self.pause(10) @@ -236,84 +106,3 @@ class AX: record = json.loads(result_path.read_text()) record["ax_attempts"] = attempts result_path.write_text(json.dumps(record)) - - def _execute_attempt(self, checkout, output, pin, pin_id, attempt, attempts): - digest = hashlib.sha256(f"{self.run}/{pin_id}/attempt-{attempt}".encode()).hexdigest()[:12] - slug = re.sub(r"[^a-z0-9]+", "-", str(pin_id).lower()).strip("-")[:22] - name = f"ousast-engine-{slug or 'pin'}-{digest}" - key_id = hashlib.sha256(str(pin_id).encode()).hexdigest() - prefix = f"engine-queue/{self.run}/{key_id}/attempt-{attempt}/" - keys = [prefix + suffix for suffix in (f"source{PRESIGNED_ARCHIVE_SUFFIX}", "questions.json", f"result{PRESIGNED_ARCHIVE_SUFFIX}")] - expires = timedelta(seconds=self.args.deadline + 600) - worker_ip = None - submitted = False - started = self.clock() - phase = "" - try: - self.store._put(keys[0], pack(checkout)) - self.store._put(keys[1], json.dumps(pin).encode()) - urls = { - "SOURCE_URL": self.store.presign_get(keys[0], expires), - "QUESTIONS_URL": self.store.presign_get(keys[1], expires), - "RESULT_URL": self.store.presign_put(keys[2], expires), - } - if any(public_endpoint(url) != self.host for url in urls.values()): - raise EgressError("AX presign hostname must match S3_ENDPOINT for the store egress policy") - manifest = task_manifest(name, self.args, urls) - end = self.clock() + self.args.deadline - # An apply timeout can still have created the task; cleanup must cover it. - submitted = True - self.ax("apply", "-f", "-", manifest=manifest, timeout=min(30, self.args.deadline)) - while self.clock() < end: - result = bounded_read(lambda: self.store._get(keys[2]), end - self.clock()) - if self.clock() >= end: - return None, worker_ip - phase = "" - try: - state = self.ax("get", "tasks", timeout=min(30, end - self.clock())).stdout - phase, ip = task_state(state, name) - worker_ip = ip or worker_ip - except (ValueError, OSError, subprocess.SubprocessError): - # The object is authoritative even when AX status is temporarily unavailable. - pass - if self.clock() >= end: - return None, worker_ip - if result is not None: - self.receive(result[0], output, pin) - return subprocess.CompletedProcess([], 0, "", ""), worker_ip - if phase.lower() in {"failed", "missing"}: - # The worker may have uploaded while we were reading AX status. - result = bounded_read(lambda: self.store._get(keys[2]), end - self.clock()) - if self.clock() >= end: - return None, worker_ip - if result is not None: - self.receive(result[0], output, pin) - return subprocess.CompletedProcess([], 0, "", ""), worker_ip - raise SandboxFailure - self.pause(min(2, max(0, end - self.clock()))) - return None, worker_ip - except (subprocess.TimeoutExpired, TimeoutError): - return None, worker_ip - except SandboxFailure: - raise - except Exception as exc: - # Never expose exception strings which may contain object-scoped bearer URLs. - detail = str(exc) if isinstance(exc, EgressError) else type(exc).__name__ - return subprocess.CompletedProcess([], 1, "", "AX transport failed: " + detail), worker_ip - finally: - attempts.append({"name": name, "phase": phase, "seconds": self.clock() - started}) - errors = [] - if submitted: - try: - self.ax("delete", "task", name) - except Exception as exc: - # A successful list already confirmed a disappeared task is gone. - if phase != "Missing": - errors.append("task: " + type(exc).__name__) - for key in keys: - try: - self.store._delete(key) - except Exception as exc: - errors.append("object: " + type(exc).__name__) - if errors: - raise RuntimeError("AX cleanup failed; stop dispatching: " + ", ".join(errors)) diff --git a/benchmarks/measurements/2026-10-04-unseen-06-scorer/record.json b/benchmarks/measurements/2026-10-04-unseen-06-scorer/record.json new file mode 100644 index 0000000..d17f337 --- /dev/null +++ b/benchmarks/measurements/2026-10-04-unseen-06-scorer/record.json @@ -0,0 +1,41 @@ +{ + "commit": "48ae918dec25119142ae781607ba99306ae5ec17", + "fixture_bytes": 15374, + "fixture_sha256": "dd3c774a47d553105ed85b9cea8810736c233f870e5a015559a1524d5db5a14d", + "implementation_sha256": { + "benchmarks/ax/batch.py": "37aecac93beb25c02826e81eacaf064fbebb447a998fe139f6d94a07035b8685", + "benchmarks/unseen/replay.py": "32efb3e0b9e9d247f2780b5f3ceeb01b4c2cbda3656c75a741f9925621cfdfbc", + "benchmarks/unseen/replay_entry.py": "994d8e239eb505dc2c5bad4cdb8b5d45967a8a2aa9ef48fb4c2f20ba45736280", + "benchmarks/unseen/score.py": "56e0a7b34482b5d84df39824b8ee447c283e35bd932f38cfe692cf84cb52cef7" + }, + "network_proofs": { + "4.4": "coordinator_pending", + "5.4": "coordinator_pending" + }, + "scope": "offline synthetic scorer only", + "score_command_exit": 0, + "scorer_output_sha256": "77f154eedb6a4acd098ab63a0b17434291f0d226dabe8ac66f2e0b362154d002", + "scorer_test_exit": 0, + "scorer_tests": 14, + "synthetic_records": 20, + "uncommitted": true, + "validation": { + "generic_dispatcher_tests": { + "exit_code": 0, + "passed": 5 + }, + "mypy": 0, + "proof_and_scorer_tests": { + "exit_code": 0, + "passed": 16 + }, + "requested_pytest": { + "exit_code": 0, + "passed": 279, + "skipped": 8 + }, + "reserved_repository_guard": 0, + "ruff_check": 0, + "ruff_format_check": 0 + } +} diff --git a/benchmarks/unseen/README.md b/benchmarks/unseen/README.md index 9274f76..9af5c38 100644 --- a/benchmarks/unseen/README.md +++ b/benchmarks/unseen/README.md @@ -197,3 +197,95 @@ are only a proxy. A working planning range is **8–48 hours**, potentially long large histories: every first-parent commit is inspected and lazy blobs can dominate. These are assumptions, not evidence from a network run. The command reports actual counts and wall time; task 2.5 establishes the measured estimate. + +## Dispatcher, replay and scoring (groups 4–6) + +`benchmarks.ax.batch` owns Task transport and the shared scheduler. New workloads supply +`Item` input bytes/commands and a `Workload` image, URL allowlist, deadline and result +validator. Two AX slots and one VM slot are the default; sandbox death gets one fresh +Task before VM fallback, OOM goes straight to VM, and instrument failures get one rerun. +Cleanup failure prevents further launches. Create `STOP` in the output directory to stop +launching; remove it and repeat the command to resume matching input digests. The legacy +engine adapter retains its `questions.json` object name for compatibility; replay inputs +and outputs use `.dat`. Both use the same object polling and cleanup implementation. + +Replay reads `pool-pN.toml`, verifies its adjacent `freeze-pN.json` canonical digest, +selects `--slice`, and resolves private pointers through `private/pool-pN.toml` (or +`--private-manifest`). It never reads a draft. `--dry-run` prints only counts and digests. +The five-change proof uses a separate frozen JSON manifest from this project's own +source-changing history. Raw output must be outside the repository. + +Each output root contains `inputs.json` and digest-keyed records and artifact directories. +The scorer accepts that root directly, joins change labels, and includes missing or stale +results as instrument failures. It refuses a slice above 5% failure after reruns. Example: + +```sh +python -m benchmarks.unseen.score "$HOME/ousast-results/unseen/p1/1/default" \ + --out "$HOME/ousast-results/unseen/p1/1/default-score.json" +``` + +For paired binary arms add `--against BASELINE_ROOT --slice-id p1/2`; `--informed` forces +`unseen.gate=not_run`. The later experiment integration owns ledger verification; callers +of `gate` must pass that ledger decision. Score-bearing arms use `budgets` with fixed +thresholds and `gate(..., binary=False, threshold=..., threshold_fixed=True)`. In-slice +thresholds are exploratory and cannot qualify adoption. JVM wall time is explicitly the +sum of the hook artifact's scan build/query times, not a separately observed process timer. + +### Coordinator task 4.4 + +Run from the merged repository root. Set `ENGINE_IMAGE` to the **engine** digest printed +by Engine image CI, `ENGINE_UNITS` and `ENGINE_EXAMPLES` to the frozen input files for the +two stored pins, `ENGINE_BASELINE` to their stored record directory, and `AX_KUBECONFIG` +to the operator's kube-ax kubeconfig. Host store credentials stay on the VM; use the public +HTTPS `S3_ENDPOINT=https://files.because-security.com` and the operator's `OUSAST_MEMORY`. +Use a fresh output directory, so the smoke executes instead of resuming prior records. + +```sh +export PATH=/home/mc/Source/OpenUltraSAST/.venv/bin:$PATH +export PYTHONPATH=$PWD/src +export S3_ENDPOINT=https://files.because-security.com +python benchmarks/learn/engine_trace.py \ + --units "$ENGINE_UNITS" --examples "$ENGINE_EXAMPLES" \ + --executor ax --parallel 2 --limit 2 --image "$ENGINE_IMAGE" \ + --kubeconfig "$AX_KUBECONFIG" --out "$HOME/ousast-results/unseen/proof-04" +python -m benchmarks.unseen.proof engine --image "$ENGINE_IMAGE" \ + --records "$HOME/ousast-results/unseen/proof-04" --baseline "$ENGINE_BASELINE" \ + --out "benchmarks/measurements/$(date +%F)-unseen-04-dispatcher/record.json" +``` + +### Coordinator task 5.4 + +Merge before running this proof. `.github/workflows/engine-image.yml` builds the `engine` +and `replay` targets on push to main; it publishes `ousast-engine-task` and +`ousast-replay-task` separately and prints their digests. Set `REPLAY_IMAGE` to the latter +full `ghcr.io/...@sha256:...` reference. No manual image build is needed. Set +`KIND_KUBECONFIG` and `AX_KUBECONFIG` to the corresponding AX kubeconfigs; the kind lane +uses the identical Task API/manifest, never a Kubernetes Job. + +```sh +export PATH=/home/mc/Source/OpenUltraSAST/.venv/bin:$PATH +export PYTHONPATH=$PWD/src +export S3_ENDPOINT=https://files.because-security.com +python -m benchmarks.unseen.replay --proof-history \ + --manifest "$HOME/ousast-results/unseen/proof-05-inputs.json" +python -m benchmarks.unseen.replay --manifest "$HOME/ousast-results/unseen/proof-05-inputs.json" \ + --image "$REPLAY_IMAGE" --lane kind --kubeconfig "$KIND_KUBECONFIG" \ + --out "$HOME/ousast-results/unseen/proof-05-kind" +python -m benchmarks.unseen.replay --manifest "$HOME/ousast-results/unseen/proof-05-inputs.json" \ + --image "$REPLAY_IMAGE" --lane ax --kubeconfig "$AX_KUBECONFIG" \ + --out "$HOME/ousast-results/unseen/proof-05-ax" +python -m benchmarks.unseen.replay --manifest "$HOME/ousast-results/unseen/proof-05-inputs.json" \ + --image "$REPLAY_IMAGE" --lane docker \ + --out "$HOME/ousast-results/unseen/proof-05-vm" +python -m benchmarks.unseen.proof replay --image "$REPLAY_IMAGE" \ + --kind-records "$HOME/ousast-results/unseen/proof-05-kind" \ + --ax-records "$HOME/ousast-results/unseen/proof-05-ax" \ + --vm-records "$HOME/ousast-results/unseen/proof-05-vm" \ + --out "benchmarks/measurements/$(date +%F)-unseen-05-replay-task/record.json" +``` + +The record command fails if fewer than five results exist, identities differ, bytes/JVM +proof is missing, or quick-tier findings differ. It does not convert a quick-only run +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. diff --git a/benchmarks/unseen/fixtures/scorer.json b/benchmarks/unseen/fixtures/scorer.json new file mode 100644 index 0000000..f658d99 --- /dev/null +++ b/benchmarks/unseen/fixtures/scorer.json @@ -0,0 +1,860 @@ +[ + { + "id": "0", + "repo": "0", + "kind": "ordinary", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [ + { + "cwe": "CWE-89", + "path": "app.py", + "line": 10 + } + ] + } + ] + } + }, + { + "id": "1", + "repo": "0", + "kind": "ordinary", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + }, + { + "id": "2", + "repo": "1", + "kind": "ordinary", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + }, + { + "id": "3", + "repo": "1", + "kind": "ordinary", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + }, + { + "id": "4", + "repo": "2", + "kind": "ordinary", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + }, + { + "id": "5", + "repo": "2", + "kind": "ordinary", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + }, + { + "id": "6", + "repo": "3", + "kind": "ordinary", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + }, + { + "id": "7", + "repo": "3", + "kind": "ordinary", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + }, + { + "id": "8", + "repo": "4", + "kind": "ordinary", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + }, + { + "id": "9", + "repo": "4", + "kind": "ordinary", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + }, + { + "id": "10", + "repo": "5", + "kind": "ordinary", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + }, + { + "id": "11", + "repo": "5", + "kind": "ordinary", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + }, + { + "id": "12", + "repo": "6", + "kind": "ordinary", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + }, + { + "id": "13", + "repo": "6", + "kind": "ordinary", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + }, + { + "id": "14", + "repo": "7", + "kind": "ordinary", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + }, + { + "id": "15", + "repo": "7", + "kind": "ordinary", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + }, + { + "id": "16", + "repo": "8", + "kind": "vulnerability", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [ + { + "cwe": "CWE-89", + "path": "app.py", + "line": 10 + } + ] + } + ] + } + }, + { + "id": "17", + "repo": "8", + "kind": "vulnerability", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [ + { + "cwe": "CWE-89", + "path": "app.py", + "line": 10 + } + ] + } + ] + } + }, + { + "id": "18", + "repo": "9", + "kind": "vulnerability", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + }, + { + "id": "19", + "repo": "9", + "kind": "vulnerability", + "family": "injection", + "method": "introducing", + "post_cutoff": true, + "label_check": "confirmed", + "path": "app.py", + "function": "handler", + "files": [ + { + "path": "app.py", + "head_lines": [ + [ + 10, + 12 + ] + ] + } + ], + "instrument": { + "head_verified": true, + "base_verified": true, + "changed_bytes": { + "app.py": 20 + }, + "hook_bytes_read": 20, + "hook_exit": 0 + }, + "push": { + "quick_tier": [ + { + "files": [ + "app.py" + ], + "bytes_read": 20, + "findings": [] + } + ] + } + } +] diff --git a/benchmarks/unseen/proof.py b/benchmarks/unseen/proof.py new file mode 100644 index 0000000..b5f2fb0 --- /dev/null +++ b/benchmarks/unseen/proof.py @@ -0,0 +1,128 @@ +"""Build anonymous coordinator proof records from completed dispatcher outputs.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import subprocess +from pathlib import Path + + +def digest(value): + return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest() + + +def records(root): + return [json.loads(path.read_text()) for path in sorted(root.glob("*.json")) if path.name not in {"progress.json", "inputs.json"}] + + +def replay_proof(roots): + lanes = {lane: records(root) for lane, root in roots.items()} + if any(len(rows) != 5 for rows in lanes.values()): + raise ValueError("proof requires exactly five completed changes per lane") + keys = [{row["input_digest"] for row in rows} for rows in lanes.values()] + if not all(key == keys[0] for key in keys): + raise ValueError("lane input identities differ") + quick = {} + evidence = {} + for lane, rows in lanes.items(): + quick[lane] = {} + evidence[lane] = [] + for row in rows: + inst = row.get("instrument", {}) + if ( + row.get("status") not in {"quick_only", "engine_covered"} + or not inst.get("hook_bytes_read") + or inst.get("suspect_fast") + or inst.get("jvm_wall_seconds", 0) < 5 + ): + raise ValueError("proof instrument has unread input or implausible JVM time") + quick[lane][row["input_digest"]] = digest([t.get("findings", []) for t in row["push"].get("quick_tier", [])]) + evidence[lane].append( + { + "input_digest": row["input_digest"], + "seconds": inst["seconds"], + "jvm_seconds": inst["jvm_wall_seconds"], + "bytes": inst["hook_bytes_read"], + "exit_code": inst["hook_exit"], + } + ) + equal = all(value == next(iter(quick.values())) for value in quick.values()) + if not equal: + raise ValueError("quick-tier findings differ across lanes") + return {"lanes": evidence, "quick_tier_equal": True, "quick_digests": quick, "cleanup": "completed-before-checkpoint"} + + +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] + if len(current) != 2: + raise ValueError("engine proof requires exactly two pins") + evidence = [] + + def stable(record): + return [ + {key: u.get(key) for key in ("unit", "status", "questions_asked", "questions_completed", "witness_rows", "traces")} + for u in record["units"] + ] + + for row in current: + key = digest([row["repo"], row["pin"]]) + previous = old[key] + instruments = [u.get("instrument", {}) for u in row["units"]] + if ( + not row.get("done") + or not any(i.get("bytes", 0) > 0 for i in instruments) + or not any(stage.get("seconds", 0) >= 5 for i in instruments for stage in i.get("jvm", [])) + ): + raise ValueError("engine proof lacks bytes or realistic JVM time") + evidence.append( + { + "pin_digest": key, + "seconds": row.get("seconds"), + "record_digest": digest(stable(row)), + "baseline_digest": digest(stable(previous)), + "equal": stable(row) == stable(previous), + "instrument": [ + { + "bytes": u.get("instrument", {}).get("bytes", 0), + "cpg_bytes": u.get("instrument", {}).get("cpg_bytes", 0), + "jvm": [ + {key: stage.get(key) for key in ("seconds", "exit_code", "success", "heap_mb")} + for stage in u.get("instrument", {}).get("jvm", []) + ], + } + for u in row["units"] + ], + } + ) + return {"pins": evidence, "differences": sum(not e["equal"] for e in evidence), "cleanup": "completed-before-checkpoint"} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("kind", choices=("engine", "replay")) + parser.add_argument("--image", required=True) + parser.add_argument("--out", type=Path, required=True) + parser.add_argument("--records", type=Path) + parser.add_argument("--baseline", type=Path) + parser.add_argument("--kind-records", type=Path) + parser.add_argument("--ax-records", type=Path) + parser.add_argument("--vm-records", type=Path) + args = parser.parse_args() + result = ( + engine_proof(args.records, args.baseline) + if args.kind == "engine" + else replay_proof({"kind": args.kind_records, "ax": args.ax_records, "vm": args.vm_records}) + ) + result.update( + image=args.image, commit=subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True, text=True, check=True).stdout.strip() + ) + args.out.parent.mkdir(parents=True, exist_ok=True) + args.out.write_text(json.dumps(result, sort_keys=True, indent=2) + "\n") + + +if __name__ == "__main__": + main() diff --git a/benchmarks/unseen/replay.py b/benchmarks/unseen/replay.py new file mode 100644 index 0000000..ad1de0c --- /dev/null +++ b/benchmarks/unseen/replay.py @@ -0,0 +1,193 @@ +"""Dispatch frozen unseen changes or a five-change project-history proof.""" + +from __future__ import annotations + +import argparse +import hashlib +import io +import json +import subprocess +import tarfile +import tomllib +from pathlib import Path + +from benchmarks.ax.batch import AXLane, DockerLane, Item, Workload, dispatch, write_json +from benchmarks.unseen.score import coverage + + +def make_item(change, arm, image): + 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} + data = json.dumps(spec, sort_keys=True).encode() + digest = hashlib.sha256(data).hexdigest() + return Item( + "replay-" + identity, + {"SPEC_URL": data}, + "result.dat", + ("python3", "/app/benchmarks/unseen/replay_entry.py"), + {}, + digest, + {"SPEC_URL": "spec.dat"}, + ) + + +def validate_result(data, output): + with tarfile.open(fileobj=io.BytesIO(data)) as archive: + members = archive.getmembers() + required = {"push.json", "push.txt", "engine.json", "instrument.json"} + if {m.name for m in members} != required or any(not m.isfile() for m in members): + raise ValueError("replay result requires push, engine and instrument.json") + payload = {m.name: archive.extractfile(m).read() for m in members} + instrument = json.loads(payload["instrument.json"]) + push = json.loads(payload["push.json"]) + record = {"instrument": instrument, "push": push} + record["status"] = coverage(record) + # Transport/scheduler needs an explicit OOM signal, not a quiet failed replay. + if "OutOfMemoryError" in json.dumps(push) or "JVM out of memory" in json.dumps(push): + record["status"] = "oom" + output.mkdir(parents=True, exist_ok=True) + for name, body in payload.items(): + (output / name).write_bytes(body) + (output / "result.dat").write_bytes(data) + write_json(output / "result.json", record) + return record + + +def proof_changes(root): + """Only the operator's project history; no pool data or fetches.""" + + def git(*args): + return subprocess.run(["git", "-C", str(root), *args], capture_output=True, text=True, check=True).stdout.strip() + + changes = [] + for head in git("rev-list", "--no-merges", "--max-count=100", "HEAD", "--", "src").splitlines(): + base = git("rev-parse", head + "^") + changed = git("diff", "--name-only", base, head).splitlines() + if not any(name.startswith("src/") and name.endswith(".py") for name in changed): + continue + changes.append( + { + "id": hashlib.sha256(head.encode()).hexdigest(), + "repository_url": "https://github.com/norandom/OpenUltraSAST.git", + "repo": "project-proof", + "kind": "ordinary", + "base": base, + "head": head, + } + ) + if len(changes) == 5: + break + if len(changes) != 5: + raise ValueError("history lacks five source changes") + return changes + + +def canonical(value): + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + + +def load_slice(path, slice_id, private_path=None): + if path.suffix != ".toml": + frozen = json.loads(path.read_text()) + changes = frozen["changes"] + if frozen["sha256"] != hashlib.sha256(json.dumps(changes, sort_keys=True).encode()).hexdigest(): + raise ValueError("frozen changes digest mismatch") + return changes + if not path.name.startswith("pool-"): + raise ValueError("replay requires a frozen pool, never a draft") + pool = tomllib.loads(path.read_text()) + freeze = json.loads(path.with_name("freeze-" + path.stem.removeprefix("pool-") + ".json").read_text()) + if hashlib.sha256(canonical(pool)).hexdigest() != freeze["freeze_digest"]: + raise ValueError("frozen pool digest mismatch") + private_path = private_path or path.parent / "private" / path.name + private = {} + if private_path.exists(): + private = {row["id"]: row for row in tomllib.loads(private_path.read_text()).get("repository", [])} + changes = [] + for public in pool.get("repository", []): + if str(public["slice"]) != str(slice_id): + continue + row = public + if public.get("private"): + row = private.get(public["id"]) + if row is None: + raise ValueError("missing private pool entry") + payload = {"url": row["url"], "commits": [[c["base"], c["head"]] for c in row["changes"]]} + if hashlib.sha256(canonical(payload)).hexdigest() != public["sha256"]: + raise ValueError("private pool digest mismatch") + for change in row["changes"]: + identity = hashlib.sha256(canonical([public["id"], change["kind"], change["base"], change["head"]])).hexdigest() + changes.append( + { + **change, + "id": identity, + "repo": public["id"], + "repository_url": row["url"], + "freeze_digest": freeze["freeze_digest"], + "slice": slice_id, + } + ) + if not changes: + raise ValueError("frozen slice has no changes") + return changes + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--manifest", type=Path, required=True, help="frozen JSON: changes array plus sha256 of canonical changes JSON") + parser.add_argument("--proof-history", action="store_true", help="write a five-change own-history manifest, then exit") + parser.add_argument("--image") + parser.add_argument("--private-manifest", type=Path) + parser.add_argument("--dry-run", action="store_true") + parser.add_argument("--arm", choices=("default", "long_deadline"), default="default") + parser.add_argument("--pool", default="p1") + parser.add_argument("--slice", default="1") + parser.add_argument("--out", type=Path) + parser.add_argument("--lane", choices=("ax", "kind", "docker", "mixed"), default="mixed") + parser.add_argument("--ax-bin", default="ax") + parser.add_argument("--kubeconfig", type=Path) + parser.add_argument("--atespace", default="default") + args = parser.parse_args() + if args.proof_history: + changes = proof_changes(Path.cwd()) + write_json(args.manifest, {"changes": changes, "sha256": hashlib.sha256(json.dumps(changes, sort_keys=True).encode()).hexdigest()}) + return 0 + 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] + 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 + if args.dry_run: + print(json.dumps({"items": len(items), "input_digest": hashlib.sha256(canonical([i.input_digest for i in items])).hexdigest()})) + return 0 + if args.out.resolve().is_relative_to(Path(__file__).resolve().parents[2]): + parser.error("raw replay output must be outside the repository") + write_json( + args.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) + executors = {} + for lane in set(lanes): + transport = (DockerLane if lane == "docker" else AXLane)(args, workload) + + def execute(item, output, attempt, transport=transport): + record = {**json.loads(item.inputs["SPEC_URL"]), **transport(item, 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" + 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 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/unseen/replay_entry.py b/benchmarks/unseen/replay_entry.py new file mode 100644 index 0000000..09f9a34 --- /dev/null +++ b/benchmarks/unseen/replay_entry.py @@ -0,0 +1,219 @@ +"""Credential-free replay actor. Standard library only; PUT is completion.""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import subprocess +import sys +import tarfile +import time +import urllib.error +import urllib.request +from pathlib import Path +from urllib.parse import urlsplit + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "learn")) +from engine_task_entry import transfer + +CLUSTER_FLAGS = ( + "-Xms1700m -Xmx1700m -XX:MaxMetaspaceSize=256m -XX:ReservedCodeCacheSize=128m " + "-XX:MaxDirectMemorySize=256m -Xss512k -XX:ActiveProcessorCount=2 -XX:+UseG1GC" +) + + +class ReplayFailure(Exception): + """Locally controlled diagnostic only.""" + + +def validate_spec(spec): + parts = urlsplit(spec["repository_url"]) + if ( + parts.scheme != "https" + or not parts.hostname + or "." not in parts.hostname + or parts.username + or parts.password + or parts.query + or parts.fragment + ): + raise ReplayFailure("repository must be public HTTPS") + if parts.port not in (None, 443) or parts.hostname.endswith((".local", ".internal", ".localhost")): + raise ReplayFailure("repository must be public HTTPS") + import ipaddress + + try: + ipaddress.ip_address(parts.hostname) + except ValueError: + pass + else: + 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"]): + raise ReplayFailure("unsupported hook flags") + + +def main(*, work_root=Path("/tmp"), runner=subprocess.run, heap_profile="cluster", clock=time.monotonic): + output = work_root / "out" + output.mkdir(parents=True, exist_ok=True) + instrument = { + "status": "instrument_failure", + "head_verified": False, + "base_verified": False, + "changed_bytes": {}, + "checkout_files": 0, + "checkout_bytes": 0, + "hook_bytes_read": 0, + "jvm_wall_seconds": 0, + "jvm_time_basis": "scan build_seconds + query_seconds", + "hook_exit": None, + "timings": {}, + "heap_profile": heap_profile, + } + spec = {} + step = "GET spec" + started = clock() + + def run(command, *, timeout=180, env=None): + before = clock() + result = runner(command, capture_output=True, text=True, timeout=timeout, env=env) + instrument["timings"][step] = instrument["timings"].get(step, 0) + clock() - before + if result.returncode: + raise ReplayFailure(f"exit {result.returncode}") + return result + + try: + spec_path = work_root / "spec.dat" + transfer(os.environ["SPEC_URL"], spec_path, readiness_budget=60) + step = "validate spec" + spec = json.loads(spec_path.read_text()) + instrument["image"] = spec.get("image") + validate_spec(spec) + case = work_root / "case" + case.mkdir() + git = ["git", "-c", "core.hooksPath=/dev/null", "-c", "protocol.file.allow=never", "-C", str(case)] + step = "git init" + run([*git, "init", "-q"]) + run([*git, "remote", "add", "origin", spec["repository_url"]]) + # One total fetch budget, independent of the hook deadline. + fetch_end = clock() + 180 + for key in ("head", "base"): + step = "fetch " + key + remaining = fetch_end - clock() + if remaining <= 0: + raise ReplayFailure("fetch deadline") + run([*git, "fetch", "--no-tags", "--depth", "1", "origin", spec[key]], timeout=remaining) + step = "verify " + key + run([*git, "cat-file", "-e", spec[key] + "^{commit}"]) + instrument[key + "_verified"] = True + step = "checkout" + run([*git, "checkout", "--detach", spec["head"]]) + paths = run([*git, "ls-files", "-z"]).stdout.split("\0") + for name in filter(None, paths): + path = case / name + if path.is_file() and not path.is_symlink(): + data = path.read_bytes() + instrument["checkout_files"] += 1 + instrument["checkout_bytes"] += len(data) + if instrument["checkout_bytes"] > 500 * 1024 * 1024: + raise ReplayFailure("checkout size limit") + step = "changed files" + changed = run([*git, "diff", "--name-only", "--diff-filter=ACMRT", "-z", spec["base"], spec["head"]]).stdout.split("\0") + for name in filter(None, changed): + path = case / name + if path.is_file() and not path.is_symlink() and path.resolve().is_relative_to(case.resolve()): + instrument["changed_bytes"][name] = len(path.read_bytes()) + if not instrument["checkout_bytes"] or not sum(instrument["changed_bytes"].values()): + instrument["suspect_fast"] = True + raise ReplayFailure("no changed bytes read") + step = "pre-push" + env = { + key: value for key, value in os.environ.items() if key not in {"SPEC_URL", "RESULT_URL", "_JAVA_OPTIONS", "JDK_JAVA_OPTIONS"} + } + env["OPENULTRASAST_CPG_HEAP_MB"] = "1700" if heap_profile == "cluster" else "2560" + env["JAVA_TOOL_OPTIONS"] = CLUSTER_FLAGS if heap_profile == "cluster" else "-Xmx2560m" + command = [ + "ousast", + "pre-push", + str(case), + "--base", + spec["base"], + "--head", + spec["head"], + "--artifact", + str(output / "push.json"), + *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) + instrument["timings"][step] = clock() - before + instrument["hook_exit"] = completed.returncode + (output / "push.txt").write_text(completed.stdout) + if completed.returncode not in (0, 1): + raise ReplayFailure(f"hook exit {completed.returncode}") + step = "read push artifact" + push = json.loads((output / "push.json").read_text()) + instrument["hook_bytes_read"] = sum(t.get("bytes_read", 0) for t in push.get("quick_tier", [])) + scans = push.get("scans", []) + instrument["jvm_wall_seconds"] = sum(s.get("scan", {}).get(k, 0) for s in scans for k in ("build_seconds", "query_seconds")) + uncovered = "language_not_covered" in json.dumps(push.get("admission", {}).get("coverage_reasons", [])) + partitions = [part for scan in scans for part in scan.get("scan", {}).get("partitions", [])] + unsupported_only = ( + uncovered + and instrument["hook_bytes_read"] == 0 + and bool(partitions) + and all(part.get("status") == "unsupported" for part in partitions) + ) + instrument["suspect_fast"] = bool(scans) and not unsupported_only and instrument["jvm_wall_seconds"] < 5 + if instrument["suspect_fast"] or (instrument["hook_bytes_read"] <= 0 and not uncovered): + instrument["suspect_fast"] = True + raise ReplayFailure("suspect_fast or unread hook input") + engine = { + "repository": spec["repository_url"], + "revisions": {key: spec[key] for key in ("base", "head")}, + "scans": scans, + "admission": push.get("admission", {}), + } + (output / "engine.json").write_text(json.dumps(engine)) + instrument["status"] = "ok" + except Exception as exc: + detail = ( + f"HTTP {exc.code}" + if isinstance(exc, urllib.error.HTTPError) + else str(exc) + if isinstance(exc, ReplayFailure) + else type(exc).__name__ + ) + instrument["failure"] = {"step": step, "status": detail} + # A crash artifact must never be mistaken for a completed scan. + if not (output / "push.json").exists(): + (output / "push.json").write_text("{}") + if not (output / "push.txt").exists(): + (output / "push.txt").write_text("") + if not (output / "engine.json").exists(): + (output / "engine.json").write_text("{}") + instrument["seconds"] = clock() - started + (output / "instrument.json").write_text(json.dumps(instrument)) + try: + archive = work_root / "result.dat" + with tarfile.open(archive, "w") as tar: + for name in ("push.json", "push.txt", "engine.json", "instrument.json"): + tar.add(output / name, arcname=name) + request = urllib.request.Request( + os.environ["RESULT_URL"], data=archive.read_bytes(), method="PUT", headers={"Content-Type": "application/x-tar"} + ) + transfer(request, readiness_budget=60) + except Exception as exc: + detail = f"HTTP {exc.code}" if isinstance(exc, urllib.error.HTTPError) else type(exc).__name__ + print("PUT result: " + detail, file=sys.stderr) + return 1 + return int(instrument["status"] != "ok") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--heap-profile", choices=("cluster", "vm"), default="cluster") + raise SystemExit(main(heap_profile=parser.parse_args().heap_profile)) diff --git a/benchmarks/unseen/score.py b/benchmarks/unseen/score.py new file mode 100644 index 0000000..f0116bf --- /dev/null +++ b/benchmarks/unseen/score.py @@ -0,0 +1,308 @@ +"""Push-level unseen evaluation, explicit coverage and paired adoption evidence.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import random +from collections import Counter, defaultdict +from pathlib import Path + +from openultrasast.model.taxonomy import load_families +from openultrasast.push.report import CAPABILITY_VETOES + +STATES = ("instrument_failure", "unanalysable", "quick_only", "engine_covered") + + +def coverage(record): + instrument = record.get("instrument", {}) + push = record.get("push") + if ( + not isinstance(push, dict) + or not instrument.get("head_verified") + or not instrument.get("base_verified") + or instrument.get("suspect_fast") + or instrument.get("failure") + or instrument.get("hook_exit") not in (0, 1) + or sum(instrument.get("changed_bytes", {}).values()) <= 0 + ): + return "instrument_failure" + if "language_not_covered" in json.dumps([push.get("skipped", []), push.get("admission", {}).get("coverage_reasons", [])]) and not any( + tier.get("bytes_read", 0) for tier in push.get("quick_tier", []) + ): + return "unanalysable" + if instrument.get("hook_bytes_read", 0) <= 0: + return "instrument_failure" + changed = set(instrument["changed_bytes"]) + if any( + outcome.get("status") == "completed" and outcome.get("identity", {}).get("path") in changed + for scan in push.get("scans", []) + if scan.get("side") == "head" + for outcome in scan.get("scan", {}).get("question_outcomes", []) + ): + return "engine_covered" + return "quick_only" + + +def findings(record): + push = record.get("push", {}) + admission = push.get("admission", {}) + for defect in admission.get("defects", []): + for path, line in defect.get("locations", []): + yield "alerts", {**defect, "path": path, "line": line} + seen = set() + for disposition in admission.get("dispositions", []): + delta = disposition.get("candidate", {}).get("delta", {}) + operation = delta.get("head_operation") + if ( + disposition.get("admitted") + or delta.get("novelty") not in {"new", "worsened"} + or not set(disposition.get("reasons", [])) <= CAPABILITY_VETOES + or operation is None + or delta.get("witness") is None + or delta.get("defect_id") in seen + ): + continue + seen.add(delta.get("defect_id")) + yield "advisory", {**operation, "family": delta.get("family")} + taxonomy = load_families() + for tier in push.get("quick_tier", []): + for finding in tier.get("findings", []): + family = taxonomy.family_of_cwe(finding.get("cwe") or "") + yield "quick", {**finding, "family": family.id if family else None} + + +def location_match(record, finding): + for file in record.get("files", []): + if finding.get("path") != file["path"]: + continue + line = finding.get("line", 0) + if any(start - 10 <= line <= end + 10 for start, end in file.get("head_lines", [])): + return True + if record.get("family") != "config_secrets" and file["path"] == record.get("path", file["path"]): + function = record.get("function") + if function and function != "" and finding.get("function") == function: + return True + span = record.get("function_lines") or file.get("function_lines") + if span and span[0] <= line <= span[1]: + return True + return False + + +def catches(record): + return sorted( + { + stream + for stream, finding in findings(record) + if finding.get("family") == record.get("family") and location_match(record, finding) + } + ) + + +def false_alarm(record, *, alerts_only=False): + return any(not alerts_only or stream == "alerts" for stream, _ in findings(record)) + + +def wilson(successes, n): + if n == 0: + return [None, None] + p, z = successes / n, 1.96 + centre = (p + z * z / (2 * n)) / (1 + z * z / n) + half = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / (1 + z * z / n) + return [max(0, centre - half), min(1, centre + half)] + + +def required_n(p, width): + for n in range(1, 1000000): + low, high = wilson(p * n, n) + if max(p - low, high - p) <= width: + return n + raise ValueError("requested precision exceeds sizing range") + + +def rate(values): + n, k = len(values), sum(values) + return {"n": n, "count": k, "rate": k / n if n else None, "wilson95": wilson(k, n)} + + +def cluster_interval(rows, *, seed=0, draws=2000): + clusters = defaultdict(list) + for repo, value in rows: + clusters[repo].append(value) + if not clusters: + return [None, None] + rng = random.Random(seed) + groups = list(clusters.values()) + samples = [] + for _ in range(draws): + values = [value for group in rng.choices(groups, k=len(groups)) for value in group] + samples.append(sum(values) / len(values)) + samples.sort() + return [samples[int(0.025 * (draws - 1))], samples[int(0.975 * (draws - 1))]] + + +def icc(rows): + clusters = defaultdict(list) + for repo, value in rows: + clusters[repo].append(value) + n, k = len(rows), len(clusters) + if k < 2 or n <= k: + return None + mean = sum(v for _, v in rows) / n + between = sum(len(v) * (sum(v) / len(v) - mean) ** 2 for v in clusters.values()) / (k - 1) + within = sum(sum((x - sum(v) / len(v)) ** 2 for x in v) for v in clusters.values()) / (n - k) + size = (n - sum(len(v) ** 2 for v in clusters.values()) / n) / (k - 1) + denominator = between + (size - 1) * within + return (between - within) / denominator if denominator else 0.0 + + +def mcnemar(a, b): + pairs = list(zip(a, b, strict=True)) + losses = sum(x and not y for x, y in pairs) + gains = sum(y and not x for x, y in pairs) + n = losses + gains + return min(1.0, 2 * sum(math.comb(n, i) for i in range(min(losses, gains) + 1)) / 2**n) if n else 1.0 + + +def score(records, *, seed=0): + counts = Counter(coverage(r) for r in records) + if not records or counts["instrument_failure"] / len(records) > 0.05: + raise ValueError("slice exceeds 5% instrument_failure (or is empty)") + usable = [r for r in records if coverage(r) != "instrument_failure"] + + def metrics(rows): + ordinary = [r for r in rows if r["kind"] == "ordinary"] + vulnerable = [r for r in rows if r["kind"] != "ordinary"] + clustered = [(r["repo"], int(false_alarm(r))) for r in ordinary] + return { + "catch": rate([bool(catches(r)) for r in vulnerable]), + "false_alarm": { + **rate([false_alarm(r) for r in ordinary]), + "cluster95": cluster_interval(clustered, seed=seed), + "icc": icc(clustered), + }, + "alerts_only": rate([false_alarm(r, alerts_only=True) for r in ordinary]), + } + + breakdown = {} + for key in ("family", "method", "post_cutoff"): + groups = defaultdict(list) + for r in usable: + groups[str(r.get(key, "unknown"))].append(r) + breakdown[key] = {key: metrics(rows) for key, rows in sorted(groups.items())} + breakdown["coverage"] = {state: metrics([r for r in usable if coverage(r) == state]) for state in STATES[1:]} + breakdown["stream"] = { + stream: rate([stream in catches(r) for r in usable if r["kind"] != "ordinary"]) for stream in ("alerts", "advisory", "quick") + } + return { + **metrics(usable), + "total": len(records), + "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 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"] + if binary: + fa = rate([r["prediction"] for r in ordinary]) + return { + "catch": rate([r["prediction"] for r in vulnerable]), + "false_alarm": fa, + "budgets_met": [f"{budget}%" for budget in (1, 5, 10) if fa["rate"] is not None and fa["rate"] <= budget / 100], + } + output = {} + for budget in (1, 5, 10): + key = f"{budget}%" + if thresholds is not None and key not in thresholds: + continue + if not ordinary: + output[key] = {"label": "not_run"} + continue + if thresholds is not None: + threshold = thresholds[key] + else: + candidates = sorted({r["score"] for r in rows}) + candidates.append(math.nextafter(candidates[-1], math.inf)) + threshold = next(t for t in candidates if sum(r["score"] >= t for r in ordinary) / len(ordinary) <= budget / 100) + output[key] = { + "threshold": threshold, + "label": "fixed" if thresholds is not None else "exploratory", + "catch": rate([r["score"] >= threshold for r in vulnerable]), + "false_alarm": rate([r["score"] >= threshold for r in ordinary]), + } + return output + + +def gate(a, b, *, slice_id=None, informed=False, binary=True, threshold=None, threshold_fixed=False, seed=0): + not_run = {"gate": "not_run"} + if not slice_id or informed or not a or not b or (not binary and (not threshold_fixed or threshold is None)): + return not_run + try: + score(a, seed=seed) + score(b, seed=seed) + except ValueError: + return not_run + aa, bb = {r["id"]: r for r in a}, {r["id"]: r for r in b} + if len(aa) != len(a) or len(bb) != len(b) or aa.keys() != bb.keys(): + return not_run + pairs = [ + (aa[key], bb[key]) for key in sorted(aa) if coverage(aa[key]) != "instrument_failure" and coverage(bb[key]) != "instrument_failure" + ] + if any(x["kind"] != y["kind"] or x["repo"] != y["repo"] for x, y in pairs): + return not_run + vulnerable = [(x, y) for x, y in pairs if x["kind"] != "ordinary"] + ordinary = [(x, y) for x, y in pairs if x["kind"] == "ordinary"] + if not vulnerable or not ordinary: + return not_run + ca = [bool(catches(x)) for x, _ in vulnerable] + cb = [bool(catches(y)) if binary else y["score"] >= threshold for _, y in vulnerable] + fa = [false_alarm(x) for x, _ in ordinary] + fb = [false_alarm(y) if binary else y["score"] >= threshold for _, y in ordinary] + catch_p, fa_p = mcnemar(ca, cb), mcnemar(fa, fb) + regression = (sum(cb) < sum(ca) and catch_p < 0.05) or (binary and sum(fb) > sum(fa) and fa_p < 0.05) + return { + "gate": "regression" if regression else "pass", + "slice": slice_id, + "catch": {"difference": sum(cb) / len(cb) - sum(ca) / len(ca), "mcnemar_p": catch_p}, + "false_alarm": { + "difference": sum(fb) / len(fb) - sum(fa) / len(fa), + "mcnemar_p": fa_p, + "cluster95": cluster_interval([(x["repo"], int(v) - int(u)) for (x, _), u, v in zip(ordinary, fa, fb, strict=True)], seed=seed), + }, + } + + +def load_records(path): + if path.is_file(): + return json.loads(path.read_text()) + rows = [] + for entry in json.loads((path / "inputs.json").read_text()): + result = path / (hashlib.sha256(entry["item_id"].encode()).hexdigest() + ".json") + record = json.loads(result.read_text()) if result.exists() else {} + if record.get("input_digest") != entry["input_digest"]: + record = {} + rows.append({**entry["change"], **record, "id": entry["change"]["id"]}) + return rows + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + 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("--slice-id") + parser.add_argument("--informed", action="store_true") + args = parser.parse_args() + current = load_records(args.records) + result = score(current) + result["unseen"] = gate(load_records(args.against) if args.against else [], current, slice_id=args.slice_id, informed=args.informed) + args.out.write_text(json.dumps(result, sort_keys=True, indent=2) + "\n") + + +if __name__ == "__main__": + main() diff --git a/plane/Dockerfile.engine-task b/plane/Dockerfile.engine-task index 5c059ef..23a5298 100644 --- a/plane/Dockerfile.engine-task +++ b/plane/Dockerfile.engine-task @@ -2,7 +2,7 @@ # preserve its runner as PID 1 and put the engine command in the Task manifest. FROM eclipse-temurin:21-jre-jammy AS jre -FROM ghcr.io/norandom/ax-task-runner:v0.3.1 +FROM ghcr.io/norandom/ax-task-runner:v0.3.1 AS engine LABEL org.opencontainers.image.licenses="Apache-2.0" \ org.opencontainers.image.source="https://github.com/norandom/OpenUltraSAST" @@ -50,3 +50,11 @@ RUN python3 -m venv /venv \ && /venv/bin/python -I -c 'import pathlib, runpy, sys; sys.modules.update(yaml=None, boto3=None); sys.path.insert(0, "/app/benchmarks/learn"); p = pathlib.Path("/app/benchmarks/learn/engine_trace_worker.py"); print("Worker import smoke bytes:", len(p.read_bytes())); runpy.run_path(str(p), run_name="engine_worker_import_smoke")' # Do not set ENTRYPOINT or CMD: both belong to the upstream AX runner. + +FROM engine AS replay +RUN apt-get update \ + && apt-get install -y --no-install-recommends git \ + && rm -rf /var/lib/apt/lists/* +COPY benchmarks/unseen/replay_entry.py ./benchmarks/unseen/replay_entry.py +RUN /venv/bin/pip install --no-cache-dir . \ + && /venv/bin/ousast pre-push --help > /dev/null diff --git a/src/openultrasast/plane/memory.py b/src/openultrasast/plane/memory.py index 4e1afd6..9ddebef 100644 --- a/src/openultrasast/plane/memory.py +++ b/src/openultrasast/plane/memory.py @@ -615,6 +615,8 @@ class S3Store(MemoryStore): presigned URLs. An admin configures the bucket once; opening a store verifies it (:meth:`verify_bucket`) and refuses a bucket without versioning, the ``runs/`` expiry rule or S3 Select. There is no local fallback.""" + read_only: bool = False + def __init__(self, client: ObjectClient, bucket: str, prefix: str = "", *, read_only: bool = False) -> None: self.client, self.bucket, self.prefix = client, bucket, prefix.strip("/") self.read_only = read_only diff --git a/tests/test_ax_batch.py b/tests/test_ax_batch.py new file mode 100644 index 0000000..3a877f2 --- /dev/null +++ b/tests/test_ax_batch.py @@ -0,0 +1,159 @@ +"""Offline shared dispatcher contracts.""" + +import json +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from benchmarks.ax.batch import Item, Workload, dispatch, manifest + +IMAGE = "registry/task@sha256:" + "a" * 64 + + +def item(identity="one"): + return Item(identity, {"SPEC_URL": b"{}"}, "result.dat", ("python3", "/entry.py"), {}, identity) + + +def workload(): + return Workload(IMAGE, frozenset({"SPEC_URL", "RESULT_URL"}), 900, lambda data, output: {}) + + +def test_manifest_contract(): + doc = manifest( + "ousast-engine-replay-abc", + workload(), + item(), + {"SPEC_URL": "https://files.example/spec.dat", "RESULT_URL": "https://files.example/result.dat"}, + ) + assert {v["name"] for v in doc["spec"]["env"]} == {"SPEC_URL", "RESULT_URL"} + with pytest.raises(ValueError): + manifest("bad", workload(), item(), {}) + bad = item() + bad.extra_env["AWS_SECRET_ACCESS_KEY"] = "secret" + with pytest.raises(ValueError): + manifest( + "ousast-engine-test", + workload(), + bad, + {"SPEC_URL": "https://files.example/spec.dat", "RESULT_URL": "https://files.example/result.dat"}, + ) + + +def test_dispatch_retry_fallback_resume_stop(tmp_path): + calls = [] + + def lane(it, output, attempt): + calls.append(("ax", attempt)) + return {"status": "sandbox_failure"} + + def vm(it, output, attempt): + calls.append(("docker", attempt)) + return {"status": "ok"} + + assert not dispatch([item()], workload(), tmp_path, {"ax": lane, "docker": vm}, lanes=("ax", "ax", "docker"), overflow=False) + assert [c[0] for c in calls] == ["ax", "ax", "docker"] + assert not dispatch([item()], workload(), tmp_path, {"ax": lane, "docker": vm}) + assert len(calls) == 3 + (tmp_path / "STOP").touch() + dispatch([item("two")], workload(), tmp_path, {"ax": lane, "docker": vm}) + assert len(calls) == 3 + assert json.loads((tmp_path / "progress.json").read_text())["pins_in_flight"] == 0 + + +def test_cleanup_failure_halts(tmp_path): + calls = [] + + def broken(it, output, attempt): + calls.append(it.id) + raise RuntimeError("cleanup failed") + + with pytest.raises(RuntimeError, match="cleanup"): + dispatch([item("one"), item("two")], workload(), tmp_path, {"ax": broken}, lanes=("ax",)) + assert calls == ["one"] + + +def test_ax_lane_fake_cli_and_store(tmp_path, monkeypatch): + import subprocess + from types import SimpleNamespace + + from benchmarks.ax.batch import AXLane + + calls = [] + + class Store: + objects = {} + + def _put(self, key, data): + self.objects[key] = data + + def _get(self, key): + return (b"completed", None) + + def _delete(self, key): + self.objects.pop(key, None) + + def presign_get(self, key, expiry): + assert key.endswith(".dat") + return "https://files.example/" + key + "?signature=secret" + + presign_put = presign_get + + def run(command, **kwargs): + calls.append((command, kwargs)) + if command[1] == "get": + name = json.loads(calls[0][1]["input"])["metadata"]["name"] + return subprocess.CompletedProcess(command, 0, json.dumps({"metadata": {"name": name}, "status": {"phase": "Running"}}), "") + return subprocess.CompletedProcess(command, 0, "", "") + + monkeypatch.setattr(subprocess, "run", run) + monkeypatch.setenv("S3_ENDPOINT", "https://files.example") + + def validate(data, output): + assert data == b"completed" + (output / "result.json").write_text('{"status":"ok"}') + + work = Workload(IMAGE, frozenset({"SPEC_URL", "RESULT_URL"}), 10, validate) + lane = AXLane(SimpleNamespace(ax_bin="fake-ax", kubeconfig=None, atespace="default"), work, Store()) + assert lane(item(), tmp_path, 0)["status"] == "ok" + doc = json.loads(calls[0][1]["input"]) + assert doc["metadata"]["name"].startswith("ousast-engine-") and len(doc["metadata"]["name"]) <= 49 + assert {e["name"] for e in doc["spec"]["env"]} == {"SPEC_URL", "RESULT_URL"} + assert calls[-1][0][1:3] == ["delete", "task"] + assert not lane.store.objects + + +def test_bounded_lanes_oom_and_instrument_retry(tmp_path): + import threading + import time + from collections import Counter + + active, peak, calls = Counter(), Counter(), [] + lock = threading.Lock() + + def execute(lane): + def run(it, output, attempt): + with lock: + calls.append((it.id, lane, attempt)) + active[lane] += 1 + peak[lane] = max(peak[lane], active[lane]) + time.sleep(0.01) + with lock: + active[lane] -= 1 + return {"status": "oom" if lane == "ax" else "ok"} + + return run + + assert not dispatch([item(str(i)) for i in range(9)], workload(), tmp_path, {"ax": execute("ax"), "docker": execute("docker")}) + assert peak["ax"] == 2 and peak["docker"] == 1 + assert all(attempt == 0 for _, _, attempt in calls) + attempts = [] + + def failed(it, output, attempt): + attempts.append(attempt) + return {"status": "instrument_failure"} + + assert dispatch([item("fail")], workload(), tmp_path, {"ax": failed}, lanes=("ax",)) + assert attempts == [0, 1] diff --git a/tests/test_unseen_image.py b/tests/test_unseen_image.py new file mode 100644 index 0000000..2caf81e --- /dev/null +++ b/tests/test_unseen_image.py @@ -0,0 +1,16 @@ +"""Static image contract; no Docker invocation.""" + +import hashlib +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +ENGINE_SHA = "06ec67d411cd7abbfcd0735b1c2a70e4f3eb65ddcfeef89975709c664f59ad42" + + +def test_image_stages(): + text = (ROOT / "plane/Dockerfile.engine-task").read_text() + engine, replay = text.split("FROM engine AS replay", 1) + original = engine.replace("v0.3.1 AS engine", "v0.3.1").rstrip() + "\n" + assert hashlib.sha256(original.encode()).hexdigest() == ENGINE_SHA + assert "FROM ghcr.io/norandom/ax-task-runner:v0.3.1 AS engine" in engine + assert "git" in replay and "replay_entry.py" in replay and "ousast pre-push --help" in replay diff --git a/tests/test_unseen_proof.py b/tests/test_unseen_proof.py new file mode 100644 index 0000000..0f28edf --- /dev/null +++ b/tests/test_unseen_proof.py @@ -0,0 +1,56 @@ +"""Proof records fail closed using synthetic local results.""" + +import importlib +import json +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +p = importlib.import_module("benchmarks.unseen.proof") + + +def test_replay_proof(tmp_path): + roots = {} + for lane in ("kind", "ax", "vm"): + root = tmp_path / lane + root.mkdir() + roots[lane] = root + (root / "inputs.json").write_text("[]") + for i in range(5): + (root / f"{i}.json").write_text( + json.dumps( + { + "input_digest": str(i), + "status": "engine_covered", + "instrument": {"hook_bytes_read": 20, "jvm_wall_seconds": 8, "seconds": 30, "hook_exit": 0}, + "push": {"quick_tier": [{"findings": []}]}, + } + ) + ) + assert p.replay_proof(roots)["quick_tier_equal"] + path = roots["vm"] / "0.json" + record = json.loads(path.read_text()) + record["instrument"]["jvm_wall_seconds"] = 0 + path.write_text(json.dumps(record)) + with pytest.raises(ValueError, match="instrument"): + p.replay_proof(roots) + + +def test_engine_proof_counts_differences_without_names(tmp_path): + current, baseline = tmp_path / "current", tmp_path / "baseline" + current.mkdir() + baseline.mkdir() + for i in range(2): + record = { + "repo": "synthetic", + "pin": str(i), + "done": True, + "seconds": 20, + "units": [{"unit": str(i), "status": "path", "instrument": {"bytes": 20, "jvm": [{"seconds": 8, "success": True}]}}], + } + (current / f"{i}.json").write_text(json.dumps(record)) + (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) diff --git a/tests/test_unseen_replay.py b/tests/test_unseen_replay.py new file mode 100644 index 0000000..37a22a2 --- /dev/null +++ b/tests/test_unseen_replay.py @@ -0,0 +1,79 @@ +"""Replay workload privacy, validation and input identity.""" + +import importlib +import io +import json +import sys +import tarfile +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +r = importlib.import_module("benchmarks.unseen.replay") + + +def test_items_private_and_arm_identity(): + change = { + "id": "repo-name-secret", + "repo": "owner/private-name", + "repository_url": "https://public.example/private-name", + "base": "a" * 40, + "head": "b" * 40, + "files": [{"path": "app.py", "head_lines": [[1, 2]]}], + } + image = "registry/replay@sha256:" + "a" * 64 + a = r.make_item(change, "default", image) + b = r.make_item(change, "long_deadline", image) + assert a.input_digest != b.input_digest + assert "private-name" not in a.id and "repo-name" not in a.id + assert set(a.inputs) == {"SPEC_URL"} and not a.extra_env + spec = json.loads(a.inputs["SPEC_URL"]) + assert spec["repository_url"] == change["repository_url"] + assert json.loads(b.inputs["SPEC_URL"])["hook_flags"] == ["--deadline", "300"] + + +def test_validator_requires_instrument(tmp_path): + buffer = io.BytesIO() + with tarfile.open(fileobj=buffer, mode="w") as tar: + data = b"{}" + info = tarfile.TarInfo("push.json") + info.size = len(data) + tar.addfile(info, io.BytesIO(data)) + with pytest.raises(ValueError, match="instrument"): + r.validate_result(buffer.getvalue(), tmp_path) + + +def test_frozen_toml_slice_and_scoring_join(tmp_path): + import hashlib + + from benchmarks.unseen.score import load_records + + pool = tmp_path / "pool-p1.toml" + pool.write_text("""pool="p1" +[[repository]] +id="opaque-a" +slice=1 +url="https://public.example/repo-a" +changes=[{kind="ordinary",base="aaa",head="bbb",files=[]}] +[[repository]] +id="opaque-b" +slice=2 +url="https://public.example/repo-b" +changes=[{kind="ordinary",base="ccc",head="ddd",files=[]}] +""") + import tomllib + + freeze = hashlib.sha256(r.canonical(tomllib.loads(pool.read_text()))).hexdigest() + (tmp_path / "freeze-p1.json").write_text(json.dumps({"freeze_digest": freeze})) + changes = r.load_slice(pool, 2) + assert len(changes) == 1 and changes[0]["repo"] == "opaque-b" + (tmp_path / "inputs.json").write_text(json.dumps([{"item_id": "replay-opaque", "change": changes[0], "input_digest": "digest"}])) + assert load_records(tmp_path)[0]["kind"] == "ordinary" + assert "instrument" not in load_records(tmp_path)[0] # Missing result remains a measured failure. + result = tmp_path / (hashlib.sha256(b"replay-opaque").hexdigest() + ".json") + result.write_text(json.dumps({"input_digest": "digest", "instrument": {"head_verified": True}})) + assert load_records(tmp_path)[0]["instrument"]["head_verified"] + pool.write_text(pool.read_text() + "changed=true\n") + with pytest.raises(ValueError, match="digest"): + r.load_slice(pool, 2) diff --git a/tests/test_unseen_replay_entry.py b/tests/test_unseen_replay_entry.py new file mode 100644 index 0000000..dcda37f --- /dev/null +++ b/tests/test_unseen_replay_entry.py @@ -0,0 +1,136 @@ +"""Replay entry uses real local git and an injected transfer, never a network.""" + +import importlib +import io +import json +import subprocess +import sys +import tarfile +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +e = importlib.import_module("benchmarks.unseen.replay_entry") + + +@pytest.fixture +def local_repo(tmp_path): + repo = tmp_path / "repo" + repo.mkdir() + + def git(*args): + return subprocess.run(["git", "-C", str(repo), *args], capture_output=True, text=True, check=True).stdout.strip() + + git("init", "-q") + git("config", "user.email", "fixture@example.invalid") + git("config", "user.name", "Fixture") + (repo / "app.py").write_text("print(1)\n") + git("add", "app.py") + git("commit", "-qm", "base") + base = git("rev-parse", "HEAD") + (repo / "app.py").write_text("print(2)\n") + git("commit", "-qam", "head") + return repo, base, git("rev-parse", "HEAD") + + +@pytest.mark.parametrize("failure", [None, "missing", "fetch", "crash", "fast", "unread", "unsupported"]) +def test_entry(local_repo, tmp_path, monkeypatch, failure): + repo, base, head = local_repo + spec = { + "repository_url": "https://public.example/project.git", + "base": base, + "head": head, + "id": "opaque", + "image": "digest", + "hook_flags": [], + } + if failure == "missing": + spec["head"] = "a" * 40 + received = {} + + def transfer(request, destination=None, **kwargs): + assert kwargs["readiness_budget"] == 60 + if destination: + destination.write_text(json.dumps(spec)) + else: + with tarfile.open(fileobj=io.BytesIO(request.data)) as archive: + received.update({m.name: archive.extractfile(m).read() for m in archive.getmembers()}) + + monkeypatch.setenv("SPEC_URL", "https://files.example/spec.dat?secret") + monkeypatch.setenv("RESULT_URL", "https://files.example/result.dat?secret") + monkeypatch.setattr(e, "transfer", transfer) + real_run = subprocess.run + + def run(command, **kwargs): + if command[0] == "ousast": + assert command[command.index("--base") + 1] == base + assert kwargs["env"]["OPENULTRASAST_CPG_HEAP_MB"] == "1700" + if failure == "crash": + return subprocess.CompletedProcess(command, 2, "", "crash https://secret") + target = Path(command[command.index("--artifact") + 1]) + target.write_text( + json.dumps( + { + "quick_tier": [{"files": ["app.py"], "bytes_read": 0 if failure == "unread" else 9, "findings": []}], + "scans": [ + { + "side": "head", + "scan": {"build_seconds": 0.1 if failure == "fast" else 8, "query_seconds": 2, "question_outcomes": []}, + } + ], + } + ) + ) + if failure == "unsupported": + target.write_text( + json.dumps( + { + "quick_tier": [{"files": [], "bytes_read": 0, "findings": []}], + "admission": {"coverage_reasons": ["language_not_covered:rust:1:1"]}, + "scans": [ + { + "side": "head", + "scan": { + "build_seconds": 0.01, + "query_seconds": 0, + "partitions": [ + { + "language": "rust", + "frontend": None, + "paths": ["app.rs"], + "source_bytes": 0, + "status": "unsupported", + } + ], + }, + } + ], + } + ) + ) + return subprocess.CompletedProcess(command, 0, "Skipped: deadline\n", "") + command = [ + str(repo) if arg == spec["repository_url"] else "protocol.file.allow=always" if arg == "protocol.file.allow=never" else arg + for arg in command + ] + if failure == "fetch" and "fetch" in command: + return subprocess.CompletedProcess(command, 1, "", "fetch secret") + return real_run(command, **kwargs) + + result = e.main(work_root=tmp_path / "work", runner=run) + instrument = json.loads(received["instrument.json"]) + assert instrument["image"] == "digest" + assert b"secret" not in received["instrument.json"] + if failure == "unsupported": + from benchmarks.unseen.score import coverage + + assert result == 0 and not instrument["suspect_fast"] + assert coverage({"instrument": instrument, "push": json.loads(received["push.json"])}) == "unanalysable" + elif failure: + assert result == 1 and instrument["status"] == "instrument_failure" + else: + assert result == 0 and instrument["changed_bytes"] == {"app.py": 9} + assert instrument["head_verified"] and instrument["base_verified"] + assert instrument["jvm_wall_seconds"] == 10 + assert b"Skipped:" in received["push.txt"] diff --git a/tests/test_unseen_score.py b/tests/test_unseen_score.py new file mode 100644 index 0000000..cfcc246 --- /dev/null +++ b/tests/test_unseen_score.py @@ -0,0 +1,127 @@ +"""Synthetic push-level statistical controls; no corpus reads.""" + +import importlib +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +s = importlib.import_module("benchmarks.unseen.score") + + +def record(kind="introducing", **updates): + return { + "id": "a", + "repo": "opaque", + "kind": kind, + "family": "injection", + "function": "handler", + "files": [{"path": "app.py", "head_lines": [[20, 22]]}], + "instrument": { + "head_verified": True, + "base_verified": True, + "changed_bytes": {"app.py": 20}, + "hook_exit": 0, + "hook_bytes_read": 20, + }, + "push": {"quick_tier": [{"bytes_read": 20, "findings": []}]}, + **updates, + } + + +def test_coverage_and_failure_denominator(): + r = record() + assert s.coverage(r) == "quick_only" + r["push"]["scans"] = [{"side": "head", "scan": {"question_outcomes": [{"status": "completed", "identity": {"path": "app.py"}}]}}] + assert s.coverage(r) == "engine_covered" + r["instrument"]["suspect_fast"] = True + assert s.coverage(r) == "instrument_failure" + assert s.coverage(record(push={"skipped": ["language_not_covered:other:1:1"]})) == "unanalysable" + assert s.coverage({}) == "instrument_failure" + with pytest.raises(ValueError, match="5%"): + s.score([record(), r]) + + +@pytest.mark.parametrize( + "p,width,n", [(0.05, 0.03, 315), (0.1, 0.03, 483), (0.2, 0.03, 756), (0.1, 0.1, 62), (0.25, 0.1, 88), (0.4, 0.1, 97)] +) +def test_sample_sizes(p, width, n): + assert s.required_n(p, width) == n + + +def test_catch_and_all_stream_false_alarm(): + r = record() + r["push"]["quick_tier"][0]["findings"] = [{"cwe": "CWE-89", "path": "app.py", "line": 20}] + assert s.catches(r) == ["quick"] + r["push"]["quick_tier"][0]["findings"][0]["line"] = 100 + assert not s.catches(r) + r["kind"] = "ordinary" + assert s.false_alarm(r) + assert not s.false_alarm(r, alerts_only=True) + + +def test_location_rules(): + assert s.location_match(record(family="access_control"), {"path": "app.py", "function": "handler", "line": 200}) + assert not s.location_match(record(family="config_secrets"), {"path": "app.py", "function": "handler", "line": 200}) + assert s.location_match(record(family="config_secrets"), {"path": "app.py", "line": 30}) + assert not s.location_match(record(), {"path": "else.py", "function": "handler", "line": 20}) + + +def test_bootstrap_icc_mcnemar(): + rows = [("a", 1), ("a", 1), ("b", 0), ("b", 0)] + assert s.cluster_interval(rows, seed=7) == s.cluster_interval(rows, seed=7) + assert s.icc(rows) == pytest.approx(1) + assert s.mcnemar([True] * 6, [False] * 6) == pytest.approx(0.03125) + assert s.mcnemar([True], [True]) == 1 + + +def test_budgets(): + rows = [{"kind": "ordinary", "score": x} for x in range(100)] + [{"kind": "introducing", "score": 99}] + result = s.budgets(rows) + assert result["5%"]["threshold"] == 95 + assert result["5%"]["label"] == "exploratory" + assert s.budgets(rows, thresholds={"5%": 96})["5%"]["label"] == "fixed" + binary = s.budgets([{"kind": "ordinary", "prediction": False}, {"kind": "introducing", "prediction": True}], binary=True) + assert binary["budgets_met"] == ["1%", "5%", "10%"] + + +def test_gate(): + a = [record(kind="ordinary", id=str(i), repo=str(i)) for i in range(10)] + [record(id="v")] + b = [ + record(kind="ordinary", id=str(i), repo=str(i), push={"quick_tier": [{"findings": [{"cwe": "CWE-89", "path": "x", "line": 1}]}]}) + for i in range(10) + ] + [record(id="v")] + assert s.gate(a, b, slice_id="p1/2")["gate"] == "regression" + assert s.gate(a, b)["gate"] == "not_run" + assert s.gate(a, b, slice_id="p1/2", informed=True)["gate"] == "not_run" + assert s.gate(a, [{}], slice_id="p1/2")["gate"] == "not_run" + + +def test_real_artifact_coverage_and_location_binding(): + r = record(path="app.py", function_lines=[1, 100], files=[{"path": "other.py", "head_lines": [[200, 210]]}]) + assert not s.location_match(r, {"path": "other.py", "line": 40}) + assert ( + s.coverage( + record( + instrument={"head_verified": True, "base_verified": True, "changed_bytes": {"x": 20}, "hook_exit": 0, "hook_bytes_read": 0}, + push={"admission": {"coverage_reasons": ["language_not_covered:unknown:1:1"]}}, + ) + ) + == "unanalysable" + ) + r["push"]["quick_tier"][0]["findings"] = [{"cwe": None, "path": "other.py", "line": 200}] + assert not s.catches(r) + assert s.false_alarm(r) + + +def test_committed_synthetic_fixture(): + import json + + fixture = Path(__file__).resolve().parents[1] / "benchmarks/unseen/fixtures/scorer.json" + payload = fixture.read_bytes() + assert len(payload) > 0 + report = s.score(json.loads(payload)) + assert report["catch"]["n"] == 4 and report["catch"]["count"] == 2 + assert report["false_alarm"]["n"] == 16 and report["false_alarm"]["count"] == 1 + assert report["coverage"]["quick_only"] == 20