search-with-proof: declarative demos, brain/executor split, kube-ax constraints

Demos are data the verifier executes (fixed build recipes, runtime+path start,
HTTP/CLI steps with captures; the canary is never referable). The agent loop with
the key never runs repository code: executor tasks without a key exchange commands
and results through presigned objects. AX capacity errors are back-pressure with at
most 2 tasks in flight; caches and builds under /workspace with a 2 GiB guard;
verifiers take one presigned archive per side. Launcher fix: private /dev in bwrap.
190 tests on the host; probe 5/5 demonstrated.

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

benchmarks/ax/batch.py

@@ -7,6 +7,7 @@ import ipaddress
 import json
 import os
 import queue
+import random
 import re
 import shutil
 import subprocess
@@ -34,6 +35,29 @@ class SandboxFailure(Exception):
     """A terminal AX task without an authoritative result object."""
 
 
+class NoFreeWorkers(Exception):
+    """Capacity back-pressure, never a sandbox or instrument failure."""
+
+
+class CapacityDeadline(Exception):
+    """No worker became available before the task deadline."""
+
+
+# Shared by every workload/lane instance in this dispatcher process, including
+# long-lived search executors. The operator runs one task per worker, two workers.
+AX_SLOTS = threading.BoundedSemaphore(2)
+
+
+def apply_when_available(lane, document, end):
+    while lane.clock() < end:
+        try:
+            lane.ax("apply", "-f", "-", manifest=document, timeout=min(30, end - lane.clock()))
+            return
+        except NoFreeWorkers:
+            lane.pause(min(random.uniform(15, 30), max(0, end - lane.clock())))
+    raise CapacityDeadline("no free workers available before deadline")
+
+
 def public_endpoint(url):
     """Reject endpoints the actor's hostname-based HTTPS egress policy cannot reach."""
     parts = urlsplit(url)
@@ -143,10 +167,23 @@ class AXLane:
 
     def __call__(self, item, output, attempt):
         attempts = []
+        # Docker has its own capacity and must not hold an AX worker slot.
+        slot = not isinstance(self, DockerLane)
+        if slot and not AX_SLOTS.acquire(timeout=self.workload.deadline):
+            return {"status": "capacity_timeout", "ax_attempts": attempts}
         try:
             done, worker = self._execute_attempt(item, output, attempt, attempts)
+        except CapacityDeadline:
+            return {"status": "capacity_timeout", "ax_attempts": attempts}
         except SandboxFailure:
             return {"status": "sandbox_failure", "ax_attempts": attempts}
+        except RuntimeError:
+            # Teardown did not prove the worker free; retain its reservation.
+            slot = False
+            raise
+        finally:
+            if slot:
+                AX_SLOTS.release()
         if done is None or done.returncode:
             return {"status": "instrument_failure", "ax_attempts": attempts}
         record = json.loads((output / "result.json").read_text())
@@ -165,6 +202,8 @@ class AXLane:
             timeout=max(0.01, timeout),
         )
         if done.returncode:
+            if args[0] == "apply" and "no free workers available" in (done.stdout + done.stderr).lower():
+                raise NoFreeWorkers
             # AX errors may echo manifest URLs; report only verb and exit code.
             raise ValueError(f"ax {args[0]} failed (exit {done.returncode})")
         return done
@@ -194,7 +233,11 @@ class AXLane:
             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))
+            try:
+                apply_when_available(self, document, end)
+            except CapacityDeadline:
+                submitted = False  # All applies were explicitly rejected.
+                raise
             while self.clock() < end:
                 result = bounded_read(lambda: self.store._get(keys[-1]), end - self.clock())
                 if self.clock() >= end:
@@ -225,7 +268,7 @@ class AXLane:
             return None, worker_ip
         except (subprocess.TimeoutExpired, TimeoutError):
             return None, worker_ip
-        except SandboxFailure:
+        except (SandboxFailure, CapacityDeadline):
             raise
         except Exception as exc:
             # Never expose exception strings which may contain object-scoped bearer URLs.
@@ -282,6 +325,7 @@ def manifest(name, workload, item, urls, atespace="default", *, validate_name=Tr
         raise ValueError("unexpected URL environment names")
     # Fixed names and semantic values, never inherited environment or credentials.
     patterns = {
+        "OUSAST_SCRATCH_BYTES": r"[1-9][0-9]*",
         "DEADLINE": r"[0-9]+(?:\.[0-9]+)?",
         "QUESTION_DEADLINE": r"[0-9]+(?:\.[0-9]+)?",
         "SEARCH_STEP": r"reason|explore|verify",
@@ -542,3 +586,103 @@ class DockerLane(AXLane):
                 )
             done.stdout = json.dumps({"metadata": {"name": self._docker_name}, "status": {"phase": phase}})
         return done
+
+
+class SearchExecutorTask:
+    """Host-owned executor lifecycle using the shared AX/presigning transport.
+
+    Supply a bounded tar archive of the checkout. The brain receives only the
+    command PUT/result GET pair; the task receives their opposite permissions.
+    Closing the context deletes the task even after an uncertain command.
+    """
+
+    def __init__(self, lane, checkout_archive, *, deadline=900):
+        if not 0 < deadline <= 3600 or len(checkout_archive) > 128 * 1024**2:
+            raise ValueError("executor task input limit")
+        self.lane, self.archive, self.deadline = lane, checkout_archive, deadline
+        token = uuid.uuid4().hex[:20]
+        self.name = "ousast-engine-search-exec-" + token
+        self.keys = [f"engine-queue/search-exec/{token}/{name}" for name in ("repo.tar", "command.json", "result.json", "prepared.tar.gz")]
+        self.submitted = False
+        self.slot = False
+
+    def __enter__(self):
+        from openultrasast.search.executor import ObjectStoreExecutor
+
+        self.end = time.monotonic() + self.deadline
+        if not AX_SLOTS.acquire(timeout=self.deadline):
+            raise CapacityDeadline
+        self.slot = True
+        expires = timedelta(seconds=self.deadline + 600)
+        store = self.lane.store
+        try:
+            store._put(self.keys[0], self.archive)
+            urls = {
+                "REPO_URL": store.presign_get(self.keys[0], expires),
+                "COMMAND_GET_URL": store.presign_get(self.keys[1], expires),
+                "EXECUTOR_RESULT_PUT_URL": store.presign_put(self.keys[2], expires),
+                "PREPARED_PUT_URL": store.presign_put(self.keys[3], expires),
+            }
+            command_put = store.presign_put(self.keys[1], expires)
+            result_get = store.presign_get(self.keys[2], expires)
+            if any(public_endpoint(url) != self.lane.host for url in [*urls.values(), command_put, result_get]):
+                raise EgressError("executor presign hostname must match store egress policy")
+            work = Workload(self.lane.workload.image, frozenset(urls), self.deadline, None)
+            item = Item(
+                self.name,
+                {},
+                "result.json",
+                (
+                    "python3",
+                    "-m",
+                    "openultrasast.search.executor",
+                    "--repo",
+                    "/workspace/checkout",
+                    "--task-boundary",
+                    "--deadline",
+                    str(self.deadline),
+                ),
+                {},
+                "",
+            )
+            document = manifest(self.name, work, item, urls, self.lane.args.atespace)
+            self.submitted = True
+            try:
+                apply_when_available(self.lane, document, self.end)
+            except CapacityDeadline:
+                self.submitted = False
+                raise
+            self.client = ObjectStoreExecutor(command_put, result_get)
+            return self
+        except Exception:
+            self.__exit__(None, None, None)
+            raise
+
+    def prepare(self, spec):
+        """Return the executor-built single verifier input, checking its digest."""
+        remaining = min(300, self.end - time.monotonic())
+        result = self.client.submit("prepare_verification", spec, timeout_seconds=remaining)
+        if result.get("status") != "ok":
+            raise ValueError(result.get("reason", "could_not_build"))
+        stored = bounded_read(lambda: self.lane.store._get(self.keys[3]), self.end - time.monotonic())
+        if stored is None or hashlib.sha256(stored[0]).hexdigest() != result.get("sha256"):
+            raise ValueError("executor archive missing or digest mismatch")
+        return stored[0]
+
+    def __exit__(self, *exc):
+        errors = []
+        if self.submitted:
+            try:
+                self.lane.ax("delete", "task", self.name)
+            except Exception as error:
+                errors.append(type(error).__name__)
+        for key in self.keys:
+            try:
+                self.lane.store._delete(key)
+            except Exception as error:
+                errors.append(type(error).__name__)
+        if self.slot and not errors:
+            AX_SLOTS.release()
+            self.slot = False
+        if errors:
+            raise RuntimeError("executor cleanup failed: " + ", ".join(errors))

benchmarks/ax/search_verify.py

@@ -0,0 +1,113 @@
+"""Dispatch three fresh key-free AX tasks for each verification side."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+import uuid
+from pathlib import Path
+
+from benchmarks.ax.batch import AXLane, CapacityDeadline, Item, SearchExecutorTask, Workload, validate_image
+from openultrasast.search.verify import RunObservation, SideRecord
+from openultrasast.search.verify_task import MAX_RESULT_BYTES, pack_checkout
+
+
+def side_record(value):
+    if not isinstance(value, dict) or value.get("isolation_mode") != "task-boundary":
+        raise ValueError("missing task-boundary attestation")
+    fields = SideRecord.__dataclass_fields__
+    # AXLane adds transport metadata outside the task record.
+    data = {key: value[key] for key in fields if key in value}
+    if data.get("outcome") not in {"observed", "could_not_run", "could_not_build", "no_oracle", "inconclusive"}:
+        raise ValueError("invalid side outcome")
+    runs = data.get("runs")
+    if not isinstance(runs, list) or len(runs) > 1:
+        raise ValueError("task must return one repetition")
+    for run in runs:
+        if not isinstance(run, dict):
+            raise ValueError("invalid observation")
+        if type(run.get("observed")) is not bool or not isinstance(run.get("evidence"), str):
+            raise ValueError("invalid observation")
+    if data["outcome"] == "observed" and len(runs) != 1:
+        raise ValueError("missing completed observation")
+    for number in [
+        data.get("elapsed_seconds"),
+        *(data.get(k, 0) for k in ("build_seconds", "ready_seconds", "run_seconds", "scratch_peak_bytes")),
+        *(run.get("elapsed_seconds") for run in runs),
+    ]:
+        if type(number) not in (int, float) or not math.isfinite(number) or number < 0:
+            raise ValueError("invalid observation measurement")
+    if not isinstance(data.get("reason"), str):
+        raise ValueError("invalid side reason")
+    data["runs"] = tuple(RunObservation(**run) for run in runs)
+    return SideRecord(**data)
+
+
+def validate_result(data, output):
+    if len(data) > MAX_RESULT_BYTES:
+        raise ValueError("oversized side result")
+    side_record(json.loads(data))
+    output.mkdir(parents=True, exist_ok=True)
+    (output / "result.json").write_bytes(data)
+
+
+class AXSideDispatcher:
+    """Host adapter passed as verify(task_dispatcher=...); never caches runs."""
+
+    def __init__(self, args, image, output, *, lane=None, prepare=None):
+        validate_image(image)
+        self.output = Path(output)
+        self.args, self.image, self.lane = args, image, lane
+        self.prepare = prepare
+
+    def __call__(self, *, side, demo, family, timeout_seconds, fresh_task):
+        if fresh_task is not True or side.command or side.build_command or side.dependencies:
+            raise ValueError("verification accepts checkout and declarative inputs only")
+        # The trusted host selects an executor for this side. Build/package work
+        # finishes there before any verifier worker slot is acquired.
+        workload = Workload(self.image, frozenset({"VERIFY_INPUT_URL", "RESULT_URL"}), timeout_seconds * 6 + 60, validate_result)
+        lane = self.lane or AXLane(self.args, workload)
+        spec = {"demo": demo, "family": family, "timeout_seconds": timeout_seconds}
+        try:
+            if self.prepare is not None:
+                archive = self.prepare(side, spec)
+            else:
+                with SearchExecutorTask(lane, pack_checkout(side.checkout)) as executor:
+                    archive = executor.prepare(spec)
+        except (ValueError, CapacityDeadline) as exc:
+            return SideRecord("could_not_build", (), 0, str(exc), isolation_mode="task-boundary")
+        records = []
+        for _ in range(3):
+            identity = "search-verify-" + uuid.uuid4().hex
+            item = Item(
+                identity,
+                {"VERIFY_INPUT_URL": archive},
+                "result.json",
+                ("python3", "-m", "openultrasast.search.verify_task"),
+                {},
+                hashlib.sha256(archive).hexdigest(),
+            )
+            output = self.output / identity
+            output.mkdir(parents=True)
+            raw = lane(item, output, 1)
+            # Roundtrip enforces JSON types for test adapters as for remote tasks.
+            if raw.get("status") == "capacity_timeout":
+                return SideRecord("could_not_run", (), 0, "AX capacity deadline", isolation_mode="task-boundary")
+            result = side_record(json.loads(json.dumps(raw)))
+            records.append(result)
+            if result.outcome != "observed":
+                break
+        last = records[-1]
+        return SideRecord(
+            last.outcome,
+            tuple(run for r in records for run in r.runs),
+            sum(r.elapsed_seconds for r in records),
+            last.reason,
+            sum(r.build_seconds for r in records),
+            sum(r.ready_seconds for r in records),
+            sum(r.run_seconds for r in records),
+            max(r.scratch_peak_bytes for r in records),
+            isolation_mode="task-boundary",
+            chromium_no_sandbox=any(r.chromium_no_sandbox for r in records),
+        )