From eb05ed18da487e4efc8363fe7d26e02972f2f5c9 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Tue, 6 Oct 2026 14:50:14 +0200 Subject: [PATCH] search-with-proof: no-model demo replay (executor + verifier on a stored demo) Replays a stored demo through the executor task and the verifier lane for one side or both, with per-phase timings, so verifier changes are checked on the exact image before any paid search. Co-Authored-By: Claude Fable 5.1 --- benchmarks/ax/batch.py | 3 +- benchmarks/search/demo_replay.py | 241 +++++++++++++++++++++++++ src/openultrasast/model/audit.py | 1 + src/openultrasast/search/executor.py | 19 +- tests/test_search_demo_replay.py | 251 +++++++++++++++++++++++++++ 5 files changed, 513 insertions(+), 2 deletions(-) create mode 100644 benchmarks/search/demo_replay.py create mode 100644 tests/test_search_demo_replay.py diff --git a/benchmarks/ax/batch.py b/benchmarks/ax/batch.py index ba01e81..220d8ee 100644 --- a/benchmarks/ax/batch.py +++ b/benchmarks/ax/batch.py @@ -643,7 +643,7 @@ class DockerLane(AXLane): for env in manifest["spec"]["env"]: command += ["-e", env["name"] + "=" + env["value"]] command += [manifest["spec"]["image"], *manifest["spec"]["command"][1:]] - if "openultrasast.search.executor" not in manifest["spec"]["command"]: + if not {"openultrasast.search.executor", "openultrasast.search.verify_task"}.intersection(manifest["spec"]["command"]): command += ["--heap-profile", "vm"] elif args[0] == "get": command = ["docker", "inspect", "--format", "{{.State.Running}} {{.State.ExitCode}}", self._docker_name] @@ -759,6 +759,7 @@ class SearchExecutorTask: """Check the export manifest and every fragment before handing off the archive.""" remaining = min(300, self.end - time.monotonic()) result = self.client.submit("prepare_verification", spec, timeout_seconds=remaining) + self.preparation_result = result if result.get("status") != "ok": from openultrasast.search.task_storage import CommandFailure diff --git a/benchmarks/search/demo_replay.py b/benchmarks/search/demo_replay.py new file mode 100644 index 0000000..f0e04a4 --- /dev/null +++ b/benchmarks/search/demo_replay.py @@ -0,0 +1,241 @@ +"""Replay a stored pilot demo without a board or model; output DIR/record.json. + +Live execution acquires repositories inside executor tasks. Offline tests inject +an in-process executor and a fake lane; this CLI itself is not an offline clone. +""" + +from __future__ import annotations + +import argparse +import copy +import json +import re +import tempfile +import time +from dataclasses import asdict +from pathlib import Path + +from benchmarks.ax.batch import DockerLane +from benchmarks.search.pilot_run import ( + FAMILY_ORACLES, + AXLane, + AXSideDispatcher, + BrowserExecutor, + Pilot, + Side, + Workload, + diagnostic, + digest, + exception_reason, + load_demo, + validate_image, + validate_oracle, + validate_result, + verification_run_seconds, + verification_timeout, + verify, +) +from openultrasast.search.verify import SideRecord + + +class Replay(Pilot): + """Reuse pilot acquisition/redaction without constructing its board or worker.""" + + def __init__(self, args, root, lane, demo): + self.args, self.root, self.lane = args, root, lane + self.deadline = float("inf") + self.row = dict(repo=args.repo_url, vulnerable=args.commit, fixed=args.fixed_commit or args.commit, file=demo["start"]["path"]) + self.record = dict(tasks_submitted=0, executor_tasks=0, verify_tasks=0, input_bytes=0, failures=[]) + self.preparations = [] + + def prepare(self, revision, spec): + row = dict( + revision_digest=digest(revision), + wall_seconds_by_phase={}, + build_exit_code=None, + build_stderr_tail="", + archive_bytes=0, + archive_parts=0, + ) + self.preparations.append(row) + started = time.monotonic() + task = None + try: + with self._executor(revision) as task: + row["wall_seconds_by_phase"]["acquisition"] = time.monotonic() - started + before = time.monotonic() + try: + return task.prepare(spec) + finally: + row["wall_seconds_by_phase"]["prepare_roundtrip"] = time.monotonic() - before + except Exception as exc: + self.failure("build_export", exc) + # The dispatcher preserves build failure semantics and we retain diagnostics. + raise ValueError(self.record["failures"][-1]["reason"]) from exc + finally: + result = getattr(task, "preparation_result", {}) + row["wall_seconds_by_phase"].update(result.get("wall_seconds_by_phase", {})) + row.update( + build_exit_code=result.get("exit_code"), + build_stderr_tail=result.get("stderr", ""), + archive_bytes=result.get("archive_bytes", 0), + archive_parts=result.get("archive_parts", 0), + wall_seconds=time.monotonic() - started, + ) + + +def run_replay(args, root, *, lane=None): + started = time.monotonic() + demo = load_demo(args.demo) + timeout = verification_timeout(demo) + replay = Replay(args, root, lane, demo) + record = dict( + type="demo_replay", + repo_digest=digest(args.repo_url), + revision_digest=digest(args.commit), + family=args.family, + lane=args.lane, + both=args.both, + model_calls=0, + spend_usd=0, + ) + if args.both: + record["fixed_revision_digest"] = digest(args.fixed_commit) + if not FAMILY_ORACLES[args.family]: + result = SideRecord("no_oracle", (), 0, "no configured owned oracle") + sides = [] if args.both else [result] + else: + validate_oracle(args.family, demo["oracle"]) + workload = Workload( + args.image, + frozenset({"VERIFY_INPUT_URL", "RESULT_URL"}), + verification_run_seconds(demo, timeout) + 60, + validate_result, + kind="search-verify", + ) + lane = lane or (DockerLane if args.lane == "docker" else AXLane)(args, workload) + if isinstance(lane, DockerLane): + lane._diagnostics = [] + lane._prepare_image() + lane._prepared = True + replay.lane = copy.copy(lane) + if isinstance(lane, DockerLane): + replay.lane.ax = replay.lane._docker_command + keys = [Side(root / "affected"), Side(root / "fixed")] + pins = {keys[0].checkout: args.commit, keys[1].checkout: args.fixed_commit} + + def counted_lane(*a, **kw): + replay.admit_task() + replay.record["verify_tasks"] += 1 + return lane(*a, **kw) + + dispatcher = AXSideDispatcher( + args, args.image, root / "verify", lane=counted_lane, prepare=lambda side, spec: replay.prepare(pins[side.checkout], spec) + ) + + def dispatch(**kwargs): + before = time.monotonic() + try: + return dispatcher(**kwargs) + except Exception as exc: + replay.failure("verify", exc) + return SideRecord( + "could_not_run", (), time.monotonic() - before, replay.record["failures"][-1]["reason"], isolation_mode="task-boundary" + ) + + if args.both: + result = verify( + *keys, + args.demo, + args.family, + timeout_seconds=timeout, + oracle=demo["oracle"], + task_dispatcher=dispatch, + browser=BrowserExecutor(task_boundary=True) if demo["oracle"] == "xss" else None, + ) + sides = result.sides + else: + result = dispatch(side=keys[0], demo=demo, family=args.family, timeout_seconds=timeout, fresh_task=True) + sides = [result] + phases = {} + for preparation in replay.preparations: + for phase, seconds in preparation["wall_seconds_by_phase"].items(): + phases[phase] = phases.get(phase, 0) + seconds + for phase in ("build", "ready", "run"): + phases["verify_" + phase] = sum(getattr(s, phase + "_seconds") for s in sides) + record.update( + outcome=result.outcome, + reason=result.reason, + sides=[asdict(s) for s in sides], + preparations=replay.preparations, + **replay.record, + wall_seconds_by_phase=phases, + wall_seconds=time.monotonic() - started, + ) + # Apply the pilot's diagnostic redaction to every free-form string, including evidence. + private = tuple(str(v) for v in replay.row.values()) + (str(root), args.repo_url.removeprefix("https://github.com/")) + + def redact(value): + if isinstance(value, str): + return diagnostic(value, maximum=1500, private=private) + if isinstance(value, list): + return [redact(v) for v in value] + if isinstance(value, dict): + return {k: redact(v) for k, v in value.items()} + return value + + return redact(record) + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + for flag in ("repo-url", "commit", "image"): + parser.add_argument("--" + flag, required=True) + parser.add_argument("--demo", type=Path, required=True) + parser.add_argument("--family", choices=tuple(FAMILY_ORACLES), required=True) + parser.add_argument("--lane", choices=("ax", "docker"), required=True) + parser.add_argument("--out", type=Path, required=True) + parser.add_argument("--both", action="store_true") + parser.add_argument("--fixed-commit") + parser.add_argument("--ax-bin", default="ax") + parser.add_argument("--atespace", default="default") + parser.add_argument("--kubeconfig") + parser.add_argument("--docker-image-bytes", type=int) + args = parser.parse_args(argv) + if args.both != bool(args.fixed_commit): + parser.error("--both and --fixed-commit must be supplied together") + if not re.fullmatch(r"https://github\.com/[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", args.repo_url): + parser.error("credential-free GitHub HTTPS repository URL required") + if any(not re.fullmatch(r"[0-9a-f]{40}", pin) for pin in (args.commit, args.fixed_commit) if pin is not None): + parser.error("full lowercase commit SHA required") + try: + validate_image(args.image) + load_demo(args.demo) + except (ValueError, OSError) as exc: + parser.error(str(exc)) + if args.lane == "docker" and (args.docker_image_bytes is None or args.docker_image_bytes <= 0): + parser.error("docker requires --docker-image-bytes (unpacked image upper bound)") + args.out.mkdir(parents=True, exist_ok=True) + with tempfile.TemporaryDirectory(prefix="demo-replay-") as directory: + try: + record = run_replay(args, Path(directory)) + except Exception as exc: + record = dict( + type="demo_replay", + repo_digest=digest(args.repo_url), + revision_digest=digest(args.commit), + outcome="could_not_run", + reason=exception_reason(exc, private=(args.repo_url, args.commit, args.fixed_commit or "", directory)), + model_calls=0, + spend_usd=0, + sides=[], + ) + if args.both: + record["fixed_revision_digest"] = digest(args.fixed_commit) + (args.out / "record.json").write_text(json.dumps(record, indent=2, allow_nan=False) + "\n") + print(json.dumps(record, allow_nan=False)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/openultrasast/model/audit.py b/src/openultrasast/model/audit.py index ba44ae6..7636a04 100644 --- a/src/openultrasast/model/audit.py +++ b/src/openultrasast/model/audit.py @@ -83,6 +83,7 @@ SPEC_OWNED = { # Reachable modules that stand alone: their own CLI subcommand or external protocol, off the scan path. STANDALONE = { + "search.demo_replay": "benchmarks standalone: no-model stored demo replay; python -m benchmarks.search.demo_replay", "search.coordinator": ( "search-with-proof Req 1/2/4: injected search executor and offline board checkpoint inspector; agent worker pending" ), diff --git a/src/openultrasast/search/executor.py b/src/openultrasast/search/executor.py index 9a3ad71..d08f048 100644 --- a/src/openultrasast/search/executor.py +++ b/src/openultrasast/search/executor.py @@ -406,18 +406,24 @@ class InProcessExecutor: bundle = Path(directory) products = bundle / "products" products.mkdir() + phases = {"build": 0.0, "bundle": 0.0, "upload": 0.0} + built = {"exit_code": 0, "stderr": ""} + before = time.monotonic() if recipe["recipe"] != "none": path = self._path(recipe["arguments"][0]) argv, project = preparation_command(recipe["recipe"], path, self.repo, products) self.phase = "build" built = self._task_run(argv, self._remaining(), {"disk_bytes": task_storage.DEFAULT_SCRATCH_BYTES, **limits}, cwd=project) + phases["build"] = time.monotonic() - before if built.get("disk_limit_exceeded") or built["exit_code"] or built["timed_out"]: return { **built, "status": "could_not_build", + "wall_seconds_by_phase": phases, "phase": "build", "stderr": task_storage.diagnostic(built.get("stderr", ""), maximum=1500), } + before = time.monotonic() # Validate internal generated links before materializing them as # regular archive members. No host/external files may be exported. _tree(self.repo, generated_links=True) @@ -442,6 +448,8 @@ class InProcessExecutor: archive.seek(0) digest = hashlib.file_digest(archive, "sha256").hexdigest() archive.seek(0) + phases["bundle"] = time.monotonic() - before + before = time.monotonic() self.phase = "result upload" if self.prepared_part_urls: manifest = upload_parts( @@ -453,7 +461,16 @@ class InProcessExecutor: URLTransport().put(self.prepared_put_url, archive, self._remaining()) else: raise ValueError("multipart archive URLs required") - return {"status": "ok", "archive_bytes": size, "sha256": digest} + phases["upload"] = time.monotonic() - before + return { + "status": "ok", + "archive_bytes": size, + "sha256": digest, + "archive_parts": len(json.loads(manifest)["parts"]) if self.prepared_part_urls else 1, + "exit_code": built["exit_code"], + "stderr": task_storage.diagnostic(built.get("stderr", ""), maximum=1500), + "wall_seconds_by_phase": phases, + } def _tree_bytes(self) -> int: total = 0 diff --git a/tests/test_search_demo_replay.py b/tests/test_search_demo_replay.py new file mode 100644 index 0000000..ae5f7cd --- /dev/null +++ b/tests/test_search_demo_replay.py @@ -0,0 +1,251 @@ +"""Offline replay through real preparation, multipart export and owned oracles.""" + +import json +import shutil +import sys +import time +from dataclasses import asdict +from pathlib import Path +from types import SimpleNamespace + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from benchmarks.ax import batch +from benchmarks.search import demo_replay, pilot_run +from openultrasast.search import executor as executor_module +from openultrasast.search.executor import InProcessExecutor +from openultrasast.search.probe import materialise +from openultrasast.search.verify import Side, verify_side_task +from openultrasast.search.verify_task import extract_checkout + +IMAGE = "image@sha256:" + "a" * 64 + + +def harness(tmp_path, monkeypatch, *, failing=False): + sides, demos = materialise(tmp_path / "probe", "path") + events, objects = [], {} + + class Transport: + def put(self, url, data, timeout): + objects[url] = data if isinstance(data, bytes) else data.read() + + monkeypatch.setattr(executor_module, "URLTransport", Transport) + + class Executor(InProcessExecutor): + def _task_run(self, argv, timeout, limits, **kwargs): + if argv[:2] == ["sh", "-ec"]: + assert argv[2] == pilot_run.CLONE + source = sides[argv[-1] == "b" * 40].checkout + shutil.copytree(source, self.repo, dirs_exist_ok=True) + if failing: + (self.repo / "build.gradle").write_text("// offline failing fixture\n") + (self.repo / "gradlew").write_text("echo replay build broke >&2\nexit 7\n") + return dict(exit_code=0, stdout="cloned", stderr="") + return super()._task_run(argv, timeout, limits, **kwargs) + + class Task: + prepare = batch.SearchExecutorTask.prepare + + def __init__(self, lane, archive, deadline): + self.lane = SimpleNamespace(store=SimpleNamespace(_get=lambda key: (objects[key], {}) if key in objects else None)) + self.keys = ["unused0", "unused1", "unused2", "manifest", "part"] + self.end = time.monotonic() + deadline + + def __enter__(self): + events.append("executor") + repo = tmp_path / ("exec-" + str(len(events))) + repo.mkdir() + self.client = Executor(repo, task_boundary=True) + self.client.prepared_put_url = "manifest" + self.client.prepared_part_urls = ["part"] + return self + + def __exit__(self, *args): + events.append("deleted") + + monkeypatch.setattr(pilot_run, "SearchExecutorTask", Task) + ids = [] + + def lane(item, output, attempt): + assert events[-1] in ("deleted", "verify") + events.append("verify") + assert item.id not in ids + ids.append(item.id) + bundle = output / "bundle" + extract_checkout(item.inputs["VERIFY_INPUT_URL"], bundle) + spec = json.loads((bundle / "spec.json").read_text()) + assert spec["demo"]["build"] == {"recipe": "none", "arguments": []} + demo = output / "demo.json" + demo.write_text(json.dumps(spec["demo"])) + return asdict( + verify_side_task(Side(bundle / "checkout", products=bundle / "products"), demo, spec["family"], spec["timeout_seconds"]) + ) + + schema_path = demos["real"] / "demo.json" + if failing: + schema = json.loads(schema_path.read_text()) + schema["build"] = {"recipe": "gradle", "arguments": ["build.gradle"]} + schema_path.write_text(json.dumps(schema)) + args = SimpleNamespace( + repo_url="https://github.com/private-owner/private-project", + commit="a" * 40, + fixed_commit="b" * 40, + demo=schema_path, + family="path", + lane="ax", + image=IMAGE, + both=True, + ) + root = tmp_path / "replay" + root.mkdir() + return args, root, lane, events + + +def test_both_cli_demonstrates_probe_pair(tmp_path, monkeypatch): + args, root, lane, events = harness(tmp_path, monkeypatch) + monkeypatch.setattr(demo_replay, "AXLane", lambda *a: lane) + out = tmp_path / "out" + assert ( + demo_replay.main( + [ + "--repo-url", + args.repo_url, + "--commit", + args.commit, + "--fixed-commit", + args.fixed_commit, + "--both", + "--demo", + str(args.demo), + "--family", + "path", + "--lane", + "ax", + "--image", + IMAGE, + "--out", + str(out), + ] + ) + == 0 + ) + record = json.loads((out / "record.json").read_text()) + assert record["outcome"] == "demonstrated", record + assert events == ["executor", "deleted", "verify", "verify", "verify"] * 2 + assert [[r["observed"] for r in s["runs"]] for s in record["sides"]] == [[True] * 3, [False] * 3] + assert all(r["evidence"] for s in record["sides"] for r in s["runs"]) + assert record["input_bytes"] > 0 + assert record["model_calls"] == record["spend_usd"] == 0 + for prep in record["preparations"]: + assert prep["archive_bytes"] > 0 and prep["archive_parts"] == 1 + assert prep["build_exit_code"] == 0 + assert {"acquisition", "build", "bundle", "upload", "prepare_roundtrip"} <= prep["wall_seconds_by_phase"].keys() + assert args.repo_url not in json.dumps(record) + assert args.commit not in json.dumps(record) + assert args.fixed_commit not in json.dumps(record) + + +def test_single_build_failure_retains_stderr(tmp_path, monkeypatch): + args, root, lane, events = harness(tmp_path, monkeypatch, failing=True) + args.both, args.fixed_commit = False, None + record = demo_replay.run_replay(args, root, lane=lane) + assert record["outcome"] == "could_not_build", record + assert record["sides"][0]["outcome"] == "could_not_build" + assert record["preparations"][0]["build_exit_code"] != 0 + assert "replay build broke" in record["preparations"][0]["build_stderr_tail"] + assert events == ["executor", "deleted"] + + +@pytest.mark.parametrize("fixed", [False, True]) +def test_single_observation_is_not_a_differential(tmp_path, monkeypatch, fixed): + args, root, lane, events = harness(tmp_path, monkeypatch) + args.both = False + args.commit = args.fixed_commit if fixed else args.commit + args.fixed_commit = None + record = demo_replay.run_replay(args, root, lane=lane) + assert record["outcome"] == "observed", record + assert [r["observed"] for r in record["sides"][0]["runs"]] == [not fixed] * 3 + + +def test_no_oracle_does_not_acquire(tmp_path, monkeypatch): + args, root, lane, events = harness(tmp_path, monkeypatch) + args.family = "access_control" + record = demo_replay.run_replay(args, root, lane=lane) + assert record["outcome"] == "no_oracle" + assert not events + + +def test_both_identical_revisions_are_inconclusive(tmp_path, monkeypatch): + args, root, lane, events = harness(tmp_path, monkeypatch) + args.fixed_commit = args.commit + record = demo_replay.run_replay(args, root, lane=lane) + assert record["outcome"] == "inconclusive" + assert record["verify_tasks"] == 6 and record["tasks_submitted"] == 8 + + +def test_lane_failure_is_could_not_run(tmp_path, monkeypatch): + args, root, lane, events = harness(tmp_path, monkeypatch) + args.both, args.fixed_commit = False, None + + def broken(*args): + raise ValueError("verifier unavailable") + + record = demo_replay.run_replay(args, root, lane=broken) + assert record["outcome"] == "could_not_run" + assert "verifier unavailable" in record["reason"] + + +def test_replay_registered_as_standalone(): + from openultrasast.model.audit import audit + + rows = {r.module: r for r in audit(Path("benchmarks"))} + assert rows["search.demo_replay"].classification == "standalone_capability" + + +def test_docker_selection_uses_same_preparation_and_fresh_runs(tmp_path, monkeypatch): + args, root, fake_lane, events = harness(tmp_path, monkeypatch) + args.lane = "docker" + + class Docker(batch.DockerLane): + def __init__(self, *args): + pass + + def _prepare_image(self): + events.append("docker-image") + + def __call__(self, *args): + return fake_lane(*args) + + monkeypatch.setattr(demo_replay, "DockerLane", Docker) + record = demo_replay.run_replay(args, root) + assert record["outcome"] == "demonstrated" + assert events[0] == "docker-image" + assert record["executor_tasks"] == 2 and record["verify_tasks"] == 6 + + +def test_docker_verifier_command_has_no_engine_arguments(): + import subprocess + + commands = [] + lane = batch.DockerLane.__new__(batch.DockerLane) + lane._diagnostics = [] + + def runner(command, **kwargs): + commands.append(command) + return subprocess.CompletedProcess(command, 0, "container", "") + + lane.runner = runner + lane._docker_command( + "apply", + manifest={ + "metadata": {"name": "verifier"}, + "spec": { + "command": ["python3", "-m", "openultrasast.search.verify_task"], + "env": [], + "image": IMAGE, + }, + }, + ) + assert commands[0][-3:] == [IMAGE, "-m", "openultrasast.search.verify_task"] + assert "--heap-profile" not in commands[0]