merge(search-with-proof): demo replay tool [skip ci]

Co-Authored-By: Claude Fable 5.1 <[email protected]>

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
 

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())

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"
     ),

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

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]