diff --git a/benchmarks/ax/search_verify.py b/benchmarks/ax/search_verify.py index 791917d..e492af7 100644 --- a/benchmarks/ax/search_verify.py +++ b/benchmarks/ax/search_verify.py @@ -9,6 +9,7 @@ import uuid from pathlib import Path from benchmarks.ax.batch import AXLane, CapacityDeadline, Item, SearchExecutorTask, Workload, validate_image +from openultrasast.search.budget import verification_run_seconds from openultrasast.search.verify import RunObservation, SideRecord from openultrasast.search.verify_task import MAX_RESULT_BYTES, pack_checkout @@ -66,7 +67,13 @@ class AXSideDispatcher: 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, kind="search-verify") + workload = Workload( + self.image, + frozenset({"VERIFY_INPUT_URL", "RESULT_URL"}), + verification_run_seconds(demo, timeout_seconds) + 60, + validate_result, + kind="search-verify", + ) lane = self.lane or AXLane(self.args, workload) spec = {"demo": demo, "family": family, "timeout_seconds": timeout_seconds} try: @@ -113,4 +120,5 @@ class AXSideDispatcher: phase=last.phase, exit_code=last.exit_code, stderr=last.stderr, + stdout=last.stdout, ) diff --git a/benchmarks/search/pilot_run.py b/benchmarks/search/pilot_run.py index 17a9e73..fc16a9b 100644 --- a/benchmarks/search/pilot_run.py +++ b/benchmarks/search/pilot_run.py @@ -3,6 +3,7 @@ from __future__ import annotations import argparse +import copy import hashlib import io import json @@ -34,7 +35,7 @@ from openultrasast.plane.memory import FileStore, open_store from openultrasast.provider.openrouter import OpenRouterChatClient from openultrasast.search import _sandbox from openultrasast.search.board import Board -from openultrasast.search.budget import SearchBudget +from openultrasast.search.budget import SearchBudget, verification_run_seconds, verification_timeout from openultrasast.search.coordinator import Coordinator from openultrasast.search.demo import FAMILY_ORACLES, load_demo, validate_oracle from openultrasast.search.executor import InProcessExecutor @@ -142,6 +143,8 @@ class Pilot: command_phase=diagnostic(command.phase), exit_code=command.exit_code, stderr=diagnostic(command.stderr, maximum=1500, private=private), + stdout=diagnostic(getattr(command, "stdout", ""), maximum=1500, private=private), + ready_seconds=getattr(command, "ready_seconds", 0), ) self.record["failures"].append(entry) @@ -224,26 +227,35 @@ class Pilot: wall = self.record["wall_seconds_by_phase"] wall["build_export"] = wall.get("build_export", 0) + time.monotonic() - before - def verification(self): + def verification(self, limits=None): if not FAMILY_ORACLES[self.family]: return VerificationRecord("no_oracle", (), 0, "no configured owned oracle") demo = load_demo(self.root / "demo") validate_oracle(self.family, demo["oracle"]) + timeout = verification_timeout(demo, limits or self.worker.limits) if self.args.dry_run: sides = self.sides if self.args.side == "vulnerable" else (self.sides[1], self.sides[1]) - return verify(*sides, self.root / "demo", self.family, oracle=demo["oracle"]) + return verify(*sides, self.root / "demo", self.family, timeout_seconds=timeout, oracle=demo["oracle"]) revisions = [self.row[self.args.side], self.row["fixed"]] if self.args.verify_lane == "ax": # Side paths are opaque lookup keys; no checkout or opposite revision reaches the brain. sides = [Side(self.root / str(i)) for i in range(2)] pins = {s.checkout: pin for s, pin in zip(sides, revisions, strict=True)} parent = self + verify_lane = copy.copy(self.lane) + verify_lane.workload = Workload( + self.args.image, + frozenset({"VERIFY_INPUT_URL", "RESULT_URL"}), + verification_run_seconds(demo, timeout) + 60, + validate_result, + kind="search-verify", + ) class CountingLane: def __call__(self, *a, **kw): parent.admit_task() parent.record["verify_tasks"] += 1 - return parent.lane(*a, **kw) + return verify_lane(*a, **kw) dispatcher = AXSideDispatcher( self.args, @@ -256,6 +268,7 @@ class Pilot: *sides, self.root / "demo", self.family, + timeout_seconds=timeout, oracle=demo["oracle"], task_dispatcher=dispatcher, browser=BrowserExecutor(task_boundary=True) if demo["oracle"] == "xss" else None, @@ -263,7 +276,7 @@ class Pilot: if _sandbox.isolation_mode() != "userns": raise _sandbox.IsolationUnavailable("VM verification requires userns") sides = [] - spec = dict(demo=demo, family=self.family, timeout_seconds=5) + spec = dict(demo=demo, family=self.family, timeout_seconds=timeout) for revision in revisions: archive = self.prepare(revision, spec) bundle = self.root / ("built-" + uuid.uuid4().hex) @@ -280,6 +293,7 @@ class Pilot: *sides, self.root / "verify-demo", self.family, + timeout_seconds=timeout, oracle=demo["oracle"], browser=BrowserExecutor() if demo["oracle"] == "xss" else None, ) @@ -293,7 +307,7 @@ class Pilot: end = "execution_failure" try: if task.step == "verify": - result = self.verification() + result = self.verification(task.limits) self.verifications.append(result.outcome) self.record["outcome"] = result.outcome if self.args.side == "fixed" and result.sides: @@ -393,7 +407,7 @@ def main(argv=None): parser.add_argument("--dry-run", action="store_true") parser.add_argument("--verify-lane", choices=("vm", "ax"), default="vm") parser.add_argument("--model", default=DEFAULT_DETECTOR_MODEL) - parser.add_argument("--max-tokens", type=int, default=1024) + parser.add_argument("--max-tokens", type=int, default=4096) parser.add_argument("--usd-per-mtok-in", type=float) parser.add_argument("--usd-per-mtok-out", type=float) parser.add_argument("--image") @@ -501,7 +515,9 @@ def main(argv=None): try: store = open_store(args.board_memory) if args.board_memory else FileStore(root / "boards") if not args.dry_run: - lane = AXLane(args, Workload(args.image, frozenset({"VERIFY_INPUT_URL", "RESULT_URL"}), 900, validate_result, kind="search-verify")) + lane = AXLane( + args, Workload(args.image, frozenset({"VERIFY_INPUT_URL", "RESULT_URL"}), 900, validate_result, kind="search-verify") + ) except Exception as exc: aborted_reason = exception_reason(exc) for index in indices if aborted_reason is None else []: diff --git a/src/openultrasast/search/budget.py b/src/openultrasast/search/budget.py index 09c6a2b..ad1ac25 100644 --- a/src/openultrasast/search/budget.py +++ b/src/openultrasast/search/budget.py @@ -9,7 +9,11 @@ class SearchBudget: # Recalibrated 2026-10-05 on the first three measured searches: settled cost was ~$0.003 per model call, # but worst-case reservations (~$0.027 per call) exhausted the old $0.05 task allowance after 2-4 calls. spend_usd: float = 2.00 + # Both ceilings bound each task: $0.40 and 40 model steps. max_call_usd: float = 0.40 + max_steps: int = 40 + http_start_seconds: int = 90 + cli_start_seconds: int = 30 reason_rounds: int = 4 intents_per_round: int = 3 demonstrations: int = 3 @@ -28,6 +32,9 @@ class SearchBudget: if value < 0 or (value == 0 and name not in ("retries", "spend_usd")): raise ValueError(f"invalid {name}") if name in ( + "max_steps", + "http_start_seconds", + "cli_start_seconds", "reason_rounds", "intents_per_round", "demonstrations", @@ -37,3 +44,17 @@ class SearchBudget: "disk_bytes", ) and not isinstance(value, int): raise ValueError(f"{name} must be an integer") + + +def verification_timeout(schema: dict, limits: dict | None = None) -> int: + """Resolve the mode's start/command allowance from the search task budget.""" + key = "http_start_seconds" if schema["start"]["mode"] == "http" else "cli_start_seconds" + value = (limits or {}).get(key, getattr(SearchBudget(), key)) + if type(value) is not int or not 1 <= value <= 300: + raise ValueError("invalid verification timeout") + return value + + +def verification_run_seconds(schema: dict, timeout: int) -> int: + # One readiness operation, all declared steps, and driver shutdown/handshake. + return timeout * (1 + len(schema["steps"])) + 5 diff --git a/src/openultrasast/search/verify.py b/src/openultrasast/search/verify.py index 6f56c15..5470819 100644 --- a/src/openultrasast/search/verify.py +++ b/src/openultrasast/search/verify.py @@ -21,13 +21,13 @@ import time from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor from contextlib import suppress -from dataclasses import asdict, dataclass, field +from dataclasses import asdict, dataclass, field, replace from pathlib import Path from typing import Any, cast from ..sandbox import SandboxJob, SandboxResult from . import _sandbox, task_storage -from .budget import SearchBudget +from .budget import SearchBudget, verification_run_seconds, verification_timeout from .demo import BUILD_RECIPES, FAMILY_ORACLES, RUNTIMES, capture_output, load_demo, substitute, validate_oracle, write_start_files from .oracles import BrowserExecutor, Canary, Oracle, oracle_for from .task_storage import CommandFailure, diagnostic, exception_reason @@ -65,6 +65,7 @@ class SideRecord: phase: str = "" exit_code: int | None = None stderr: str = "" + stdout: str = "" @dataclass(frozen=True) @@ -138,16 +139,17 @@ def _tree(root: Path, *, generated_links: bool = False) -> str: # Executed as verifier-owned code inside the same network sandbox as the app. # Redirects are refused: only the configured loopback application is reachable. _HTTP_DRIVER = r""" -import http.client, importlib.util, json, os, socket, subprocess, sys, tempfile, time +import http.client, importlib.util, json, os, socket, subprocess, sys, time spec = importlib.util.spec_from_file_location('demo', '/verifier-demo.py') demo = importlib.util.module_from_spec(spec) spec.loader.exec_module(demo) config = json.load(sys.stdin) schema = config['schema'] -app_errors = tempfile.TemporaryFile() -app = subprocess.Popen(config['command'], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=app_errors) +app_errors = open("/scratch/" + config["errors"], "wb") +app_output = open("/scratch/" + config["output"], "wb") +app = subprocess.Popen(config['command'], stdin=subprocess.DEVNULL, stdout=app_output, stderr=app_errors) try: - until = time.monotonic() + config['timeout'] / 2 + until = time.monotonic() + config['timeout'] while True: if app.poll() is not None: raise SystemExit(app.returncode or 1) @@ -158,11 +160,16 @@ try: except OSError: if time.monotonic() >= until: raise RuntimeError('HTTP readiness timed out') - time.sleep(0.03) + time.sleep(0.5) + conn = http.client.HTTPConnection('127.0.0.1', schema['start']['port'], timeout=max(0.01, until - time.monotonic())) + conn.request('GET', '/') + response = conn.getresponse() # Any HTTP status establishes listening. + conn.close() with open('/scratch/' + config['ready'], 'w') as ready: ready.write('ready') + acknowledgment_deadline = time.monotonic() + 3 while not os.path.exists('/scratch/' + config['go']): - if time.monotonic() >= until + config['timeout'] / 2: + if time.monotonic() >= acknowledgment_deadline: raise RuntimeError('verifier readiness acknowledgment timed out') time.sleep(0.01) captures, outputs = {}, [] @@ -171,7 +178,7 @@ try: if not path.startswith('/') or path.startswith('//') or '\r' in path or '\n' in path: raise ValueError('invalid expanded HTTP path') headers = {key: demo.substitute(value, captures) for key, value in step.get('headers', {}).items()} - conn = http.client.HTTPConnection('127.0.0.1', schema['start']['port'], timeout=config['timeout'] / 2) + conn = http.client.HTTPConnection('127.0.0.1', schema['start']['port'], timeout=config['timeout']) conn.request(step['method'], path, body=demo.substitute(step.get('body', ''), captures).encode(), headers=headers) response = conn.getresponse() body = response.read(1048577) @@ -189,9 +196,8 @@ finally: except subprocess.TimeoutExpired: app.kill() app.wait() - app_errors.seek(max(0, app_errors.seek(0, 2) - 16384)) - sys.stderr.write(app_errors.read().decode(errors="replace")) app_errors.close() + app_output.close() """ @@ -206,16 +212,34 @@ def _http_steps( nonce: str, runner: Callable[..., SandboxResult], ready_check: Callable[[], None], + diagnostics: dict[str, Any] | None = None, ) -> list[str]: from . import demo as demo_module ready, go = ".ready-" + secrets.token_hex(12), ".go-" + secrets.token_hex(12) - request = {"schema": schema, "command": command, "timeout": job(command).timeout_seconds, "nonce": nonce, "ready": ready, "go": go} + diagnostics = diagnostics if diagnostics is not None else {} + request = { + "schema": schema, + "command": command, + "timeout": job(command).timeout_seconds, + "nonce": nonce, + "ready": ready, + "go": go, + "errors": ready + ".err", + "output": ready + ".out", + } + diagnostics.update(stdout="", stderr="") acknowledged = False + started = time.monotonic() + driver_job = replace( + job(command), + command=("/usr/bin/python3", "-I", "-c", _HTTP_DRIVER), + timeout_seconds=verification_run_seconds(schema, job(command).timeout_seconds), + ) with ThreadPoolExecutor(max_workers=1) as pool: future = pool.submit( runner, - job(("/usr/bin/python3", "-I", "-c", _HTTP_DRIVER)), + driver_job, scratch=scratch, mounts={**mounts, "/verifier-demo.py": Path(demo_module.__file__)}, env=env, @@ -223,19 +247,26 @@ def _http_steps( network=True, input_text=json.dumps(request), ) - deadline = time.monotonic() + job(command).timeout_seconds + deadline = time.monotonic() + job(command).timeout_seconds + 5 while not (scratch / ready).exists() and not future.done() and time.monotonic() < deadline: time.sleep(0.01) + diagnostics["ready_seconds"] = diagnostics.get("ready_seconds", 0) + time.monotonic() - started if (scratch / ready).exists(): ready_check() (scratch / go).touch() acknowledged = True result = future.result() + for key, suffix in (("stdout", ".out"), ("stderr", ".err")): + path = scratch / (ready + suffix) + if path.is_file(): + with path.open("rb") as stream: + stream.seek(max(0, path.stat().st_size - 16384)) + diagnostics[key] = diagnostic(stream.read().decode(errors="replace"), maximum=1500) if not acknowledged or result.exit_code or result.timed_out: raise CommandFailure( "start" if not acknowledged else "run", result.exit_code, - result.stderr, + diagnostics.get("stderr", "") + "\n" + result.stderr, "HTTP application/steps failed or readiness not acknowledged", ) outputs = json.loads(result.stdout) @@ -249,6 +280,7 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe runner = _run_task if task_boundary else _sandbox.run observations: list[RunObservation] = [] timings = dict(build=0.0, ready=0.0, run=0.0) + app_diagnostics: dict[str, Any] = {} disk_peak = 0 root: Path | None = None @@ -270,14 +302,15 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe time.monotonic() - started, diagnostic(reason, maximum=1500), timings["build"], - timings["ready"], - timings["run"], + timings["ready"] + app_diagnostics.get("ready_seconds", 0), + max(0.0, timings["run"] - app_diagnostics.get("ready_seconds", 0)), disk_peak, isolation_mode="task-boundary" if task_boundary else _sandbox.isolation_mode(), chromium_no_sandbox=bool(getattr(browser, "no_sandbox", False)), phase=phase, exit_code=result.exit_code if result is not None else None, - stderr=diagnostic(result.stderr, maximum=1500) if result is not None else "", + stderr=diagnostic(result.stderr if result is not None else app_diagnostics.get("stderr", ""), maximum=1500), + stdout=diagnostic(getattr(result, "stdout", app_diagnostics.get("stdout", "")), maximum=1500), ) try: @@ -380,6 +413,7 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe env=app_env, namespace_pid=oracle.namespace_pid, ) + app_diagnostics.update(stdout=ready.stdout, stderr=ready.stderr) if ready.exit_code or ready.timed_out: return record("could_not_run", "app readiness failed", "start", ready) if family == "ssrf" and oracle.observe()[0]: @@ -409,6 +443,7 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe canary.nonce, runner, http_ready_check, + app_diagnostics, ) for output in outputs: timed("run", oracle.capture, output) @@ -425,6 +460,7 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe namespace_pid=oracle.namespace_pid, input_text=substitute(step.get("stdin", ""), captures), ) + app_diagnostics.update(stdout=result.stdout, stderr=result.stderr) if result.timed_out: return record("inconclusive", "app attack timed out") if result.exit_code: @@ -507,7 +543,7 @@ def _run_task( ) -def verify_side_task(side: Side, demo: Path, family: str, timeout_seconds: int = 5) -> SideRecord: +def verify_side_task(side: Side, demo: Path, family: str, timeout_seconds: int | None = None) -> SideRecord: """Fresh AX task entry: one repetition, grouped by the trusted dispatcher. This explicit API is never selected as a fallback in the host verifier. @@ -517,10 +553,10 @@ def verify_side_task(side: Side, demo: Path, family: str, timeout_seconds: int = return SideRecord("no_oracle", (), 0, "no owned oracle", isolation_mode="task-boundary") validate_oracle(family, schema["oracle"]) kind = schema["oracle"] - if timeout_seconds < 1: + if timeout_seconds is not None and timeout_seconds < 1: raise ValueError("timeout must be positive") browser = BrowserExecutor(task_boundary=True) if kind == "xss" else None - return _side(side, demo, kind, timeout_seconds, browser, task_boundary=True) + return _side(side, demo, kind, timeout_seconds or verification_timeout(schema), browser, task_boundary=True) def verify( @@ -529,7 +565,7 @@ def verify( artefact: Path, family: str, *, - timeout_seconds: int = 5, + timeout_seconds: int | None = None, oracle: str | None = None, browser: BrowserExecutor | None = None, task_dispatcher: Callable[..., SideRecord] | None = None, @@ -538,7 +574,7 @@ def verify( started = time.monotonic() if family in FAMILY_ORACLES and not FAMILY_ORACLES[family]: return VerificationRecord("no_oracle", (), time.monotonic() - started, "no configured owned oracle for " + family) - if timeout_seconds < 1: + if timeout_seconds is not None and timeout_seconds < 1: raise ValueError("timeout must be positive") try: schema = load_demo(artefact) @@ -550,6 +586,7 @@ def verify( raise ValueError("oracle runtime unavailable") except (OSError, ValueError) as exc: return VerificationRecord("inconclusive", (), time.monotonic() - started, "invalid declarative demo: " + exception_reason(exc)) + timeout_seconds = timeout_seconds or verification_timeout(schema) mode = "task-boundary" if task_dispatcher is not None else _sandbox.isolation_mode() if task_dispatcher is None: try: diff --git a/src/openultrasast/search/worker.py b/src/openultrasast/search/worker.py index d4831f6..7ba467b 100644 --- a/src/openultrasast/search/worker.py +++ b/src/openultrasast/search/worker.py @@ -174,8 +174,8 @@ class Worker: demo: Path, model: str = "scripted", prices: Prices | None = None, - max_output_tokens: int = 1024, - max_steps: int = 32, + max_output_tokens: int = 4096, + max_steps: int = SearchBudget().max_steps, executor: RepositoryExecutor | None = None, run_budget: SpendBudget | None = None, ) -> None: @@ -262,6 +262,7 @@ class Worker: reservation.settle(actual) if run_reservation is not None: run_reservation.settle(actual) + self.finish_reason = raw["choices"][0].get("finish_reason") return cast(dict[str, Any], raw["choices"][0]["message"]) except Exception: if reservation._active: @@ -356,14 +357,20 @@ class Worker: self.observations = [] malformed = 0 plain_text = 0 - for index in range(self.max_steps): + shortening_retry = False + max_steps = min(self.max_steps, task.limits.get("max_steps", self.max_steps)) + for index in range(max_steps): messages[0]["content"] = ( EXPLORE_PROMPT + "\nSingle intent: " + json.dumps(task.payload.get("description", "")) - + f"\nCalls left including this one: {self.max_steps - index}. " + + f"\nCalls left including this one: {max_steps - index}. " "Finish now if only one remains; report confirmed observations or failure." ) + if index and index % 10 == 0: + messages.append( + {"role": "user", "content": f"Remaining steps: {max_steps - index}. Save a demo or finish within this budget."} + ) message = self._call(messages, TOOLS, meter) messages.append(message) calls = message.get("tool_calls", []) @@ -373,6 +380,27 @@ class Worker: return {"status": "execution_failure", "failure": "second plain-text reply: " + str(message.get("content", ""))[:300]} messages.append({"role": "user", "content": "Use a tool or call finish; plain text is not accepted."}) continue + truncated = False + for call in calls: + try: + json.loads(call["function"]["arguments"]) + except (ValueError, TypeError): + truncated = getattr(self, "finish_reason", None) == "length" + if truncated: + break + if truncated and not shortening_retry: + # Reject the whole turn before any tools execute; retry once, fully metered. + shortening_retry = True + for call in calls: + messages.append( + { + "role": "tool", + "tool_call_id": call["id"], + "content": json.dumps({"error": "Output truncated; no tools executed. Retry with a shorter call."}), + } + ) + continue + shortening_retry = False finishes = [call for call in calls if isinstance(call, dict) and call.get("function", {}).get("name") == "finish"] if finishes: if len(calls) > 1: @@ -432,7 +460,7 @@ class Worker: except (ValueError, OSError, KeyError, TypeError) as exc: result = {"error": type(exc).__name__, "reason": exception_reason(exc)} messages.append({"role": "tool", "tool_call_id": call["id"], "content": json.dumps(result)}) - raise BudgetExhausted("tool step limit exceeded", usd=meter.spent, calls=self.max_steps, end_detail="worker_steps") + raise BudgetExhausted("tool step limit exceeded", usd=meter.spent, calls=max_steps, end_detail="worker_steps") def __call__(self, task: SearchTask) -> dict[str, Any]: self.budget_exhausted = False diff --git a/tests/test_search_pilot_run.py b/tests/test_search_pilot_run.py index e257847..f4523df 100644 --- a/tests/test_search_pilot_run.py +++ b/tests/test_search_pilot_run.py @@ -112,7 +112,7 @@ def test_fixed_verification_never_acquires_vulnerable_revision(tmp_path, monkeyp def stop(archive, bundle): bundle.mkdir() - (bundle / "spec.json").write_text(json.dumps({"demo": schema, "family": "path", "timeout_seconds": 5})) + (bundle / "spec.json").write_text(json.dumps({"demo": schema, "family": "path", "timeout_seconds": 30})) monkeypatch.setattr(pilot_run, "extract_checkout", stop) monkeypatch.setattr(pilot_run, "verify", lambda *args, **kwargs: (_ for _ in ()).throw(StopVerification())) @@ -271,7 +271,7 @@ def test_fixed_effect_reporting_is_independent_of_differential(tmp_path, monkeyp pilot = make_pilot(tmp_path) side = SideRecord("observed", tuple(RunObservation(value, "owned", 0) for value in observed), 0, "completed") result = VerificationRecord("inconclusive", (side, side), 0, "no differential") - monkeypatch.setattr(pilot, "verification", lambda: result) + monkeypatch.setattr(pilot, "verification", lambda limits: result) task = SearchTask("task-1", "search-1", "verify", "", {}, asdict(SearchBudget())) assert pilot.dispatch(task)["outcome"] == "inconclusive" assert pilot.record["fixed_effect_observed"] is expected @@ -538,3 +538,58 @@ def test_pilot_executor_deletes_task_when_submit_times_out(tmp_path, monkeypatch assert calls[-1][0][:2] == ("delete", "task") assert len(deleted) == 4 assert pilot.record["executor_task_metrics"][-1]["end"] == "execution_failure" + + +def test_start_failure_retains_sanitized_logs_and_readiness(tmp_path): + from openultrasast.search.verify import SideRecord + + pilot = make_pilot(tmp_path) + side = SideRecord( + "could_not_run", + (), + 1.5, + "timeout", + ready_seconds=1.1, + phase="start", + exit_code=-9, + stdout="x" * 2000 + " boot token=hidden", + stderr="y" * 2000 + " stderr password=hidden", + ) + pilot.failure("verify", side.reason, side) + entry = pilot.record["failures"][-1] + assert entry["command_phase"] == "start" and entry["exit_code"] == -9 + assert entry["ready_seconds"] == 1.1 + assert len(entry["stdout"]) <= 1500 and len(entry["stderr"]) <= 1500 + assert "hidden" not in entry["stdout"] + entry["stderr"] + + +def test_ax_pilot_uses_verification_task_limits_for_deadline(tmp_path, monkeypatch): + from benchmarks.search import pilot_run + from openultrasast.search.probe import materialise + + pilot = make_pilot(tmp_path) + pilot.args.verify_lane = "ax" + pilot.args.image = "image@sha256:" + "a" * 64 + deadlines = [] + + class Lane: + workload = None + + def __call__(self, *args): + deadlines.append(self.workload.deadline) + + pilot.lane = Lane() + _, demos = materialise(tmp_path / "probe", "path") + schema = json.loads((demos["real"] / "demo.json").read_text()) + (tmp_path / "demo/demo.json").write_text(json.dumps(schema)) + captured = [] + + def verify(*args, **kwargs): + captured.append(kwargs["timeout_seconds"]) + kwargs["task_dispatcher"].lane(None, None, 1) + + monkeypatch.setattr(pilot_run, "verify", verify) + pilot.verification({"cli_start_seconds": 17}) + assert captured == [17] + assert deadlines == [17 * (len(schema["steps"]) + 1) + 65] + assert pilot.lane.workload is None diff --git a/tests/test_search_verify.py b/tests/test_search_verify.py index 752ad69..46b02c1 100644 --- a/tests/test_search_verify.py +++ b/tests/test_search_verify.py @@ -381,7 +381,16 @@ def test_http_startup_effect_vetoed_before_request_steps(tmp_path): with pytest.raises(ValueError, match="startup effect"): _http_steps( - {}, (), lambda command: SandboxJob("", command, tmp_path, {}, 1, 256, 128), tmp_path, {}, {}, None, "nonce", runner, check + {"steps": []}, + (), + lambda command: SandboxJob("", command, tmp_path, {}, 1, 256, 128), + tmp_path, + {}, + {}, + None, + "nonce", + runner, + check, ) assert checks == [True] @@ -463,7 +472,7 @@ def test_http_failure_keeps_exception_tail(tmp_path): traceback = "Traceback (most recent call last):\n" + " frame\n" * 200 + "FileNotFoundError: /dev/null\n" with pytest.raises(ValueError, match="FileNotFoundError: /dev/null"): _http_steps( - {}, + {"steps": []}, ("/bin/true",), lambda command: SandboxJob("", command, tmp_path, {}, 1, 256, 128), tmp_path, @@ -541,3 +550,150 @@ def test_app_requires_environment_and_data_files_on_both_sides(pair): assert result.outcome == "demonstrated", result assert all(len(side.runs) == 3 for side in result.sides) assert all(not (side.checkout / ".demo").exists() for side in sides) + + +@pytest.mark.parametrize("delay,timeout,expected", [(8, None, "observed"), (8, 1, "could_not_run")]) +def test_delayed_http_readiness_and_timeout_diagnostics(pair, monkeypatch, delay, timeout, expected): + try: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + except PermissionError: + pytest.skip("host policy prohibits localhost sockets") + sides, demo = pair + app = f"""import sys, time +from http.server import BaseHTTPRequestHandler, HTTPServer +print('x' * 2000 + ' boot stdout token=hidden', flush=True) +print('y' * 2000 + ' boot stderr password=hidden', file=sys.stderr, flush=True) +time.sleep({delay}) +class Handler(BaseHTTPRequestHandler): + def do_GET(self): + self.send_response(503); self.end_headers() + self.wfile.write(b'listening') + def log_message(self, *args): pass +HTTPServer(('127.0.0.1', {port}), Handler).serve_forever() +""" + (sides[0].checkout / "app.py").write_text(app) + + def change(value): + value["start"].update(mode="http", port=port) + value["steps"] = [{"type": "http", "method": "GET", "path": "/attack"}] + + edit(demo, change) + result = verify_side_task(sides[0], demo, "path", timeout) + assert result.outcome == expected, result + assert "boot stdout" in result.stdout and "boot stderr" in result.stderr + assert "hidden" not in result.stdout + result.stderr + assert len(result.stdout) <= 1500 and len(result.stderr) <= 1500 + if expected == "observed": + assert result.ready_seconds >= 8 + assert len(result.runs) == 1 + else: + assert result.phase == "start" and result.exit_code != 0 + assert 0.9 <= result.ready_seconds < 4 + assert not result.runs + + +def test_killed_http_driver_retains_app_logs(pair, monkeypatch): + from openultrasast.sandbox import SandboxResult + from openultrasast.search import verify as module + + sides, demo = pair + edit( + demo, + lambda value: ( + value["start"].update(mode="http", port=18080), + value.update(steps=[{"type": "http", "method": "GET", "path": "/"}]), + ), + ) + + def killed(job, **kwargs): + config = json.loads(kwargs["input_text"]) + for key in ("output", "errors"): + (kwargs["scratch"] / config[key]).write_text("app log token=hidden") + return SandboxResult(-9, "", "", True) + + monkeypatch.setattr(module, "_run_task", killed) + result = verify_side_task(sides[0], demo, "path", 1) + assert result.phase == "start" and result.exit_code == -9 + assert result.ready_seconds >= 0 + assert result.stdout == result.stderr == "app log token=[redacted]" + + +@pytest.mark.parametrize("timeout,ready", [(90, True), (1, False)]) +def test_http_driver_polls_eight_second_listener_before_get_and_steps(tmp_path, monkeypatch, capsys, timeout, ready): + import http.client + import io + import subprocess + import sys + import time + from types import SimpleNamespace + + from openultrasast.search import demo as demo_module + from openultrasast.search.verify import _HTTP_DRIVER + + tick = [0.0] + polls, requests = [], [] + monkeypatch.setattr(time, "monotonic", lambda: tick[0]) + + def sleep(seconds): + assert seconds == 0.5 + tick[0] += seconds + + monkeypatch.setattr(time, "sleep", sleep) + + def connect(address, timeout): + assert address == ("127.0.0.1", 18080) + polls.append(tick[0]) + if tick[0] < 8: + raise ConnectionRefusedError() + return SimpleNamespace(close=lambda: None) + + monkeypatch.setattr(socket, "create_connection", connect) + + class HTTP: + def __init__(self, host, port, timeout): + assert tick[0] >= 8 + + def request(self, method, path, **kwargs): + requests.append((method, path)) + + def getresponse(self): + return SimpleNamespace(status=503, read=lambda size: b"listening") + + def close(self): + pass + + monkeypatch.setattr(http.client, "HTTPConnection", HTTP) + + def app(*args, **kwargs): + kwargs["stdout"].write(b"boot stdout") + kwargs["stderr"].write(b"boot stderr") + return SimpleNamespace(poll=lambda: None, terminate=lambda: None, wait=lambda **kw: None) + + monkeypatch.setattr(subprocess, "Popen", app) + config = dict( + schema={"start": {"port": 18080}, "steps": [{"method": "GET", "path": "/attack"}]}, + command=["fake-app"], + timeout=timeout, + nonce="nonce", + ready="ready", + go="go", + errors="errors", + output="output", + ) + (tmp_path / "go").touch() + monkeypatch.setattr(sys, "stdin", io.StringIO(json.dumps(config))) + driver = _HTTP_DRIVER.replace("/verifier-demo.py", demo_module.__file__).replace("/scratch/", str(tmp_path) + "/") + if ready: + exec(driver, {}) + assert requests == [("GET", "/"), ("GET", "/attack")] + assert json.loads(capsys.readouterr().out) == ["listening"] + assert tick[0] == 8 + else: + with pytest.raises(RuntimeError, match="readiness timed out"): + exec(driver, {}) + assert not requests and not (tmp_path / "ready").exists() + assert polls == [i * 0.5 for i in range(len(polls))] + assert (tmp_path / "output").read_text() == "boot stdout" + assert (tmp_path / "errors").read_text() == "boot stderr" diff --git a/tests/test_search_verify_transport.py b/tests/test_search_verify_transport.py index cf812d7..924f8ef 100644 --- a/tests/test_search_verify_transport.py +++ b/tests/test_search_verify_transport.py @@ -242,3 +242,26 @@ def test_transport_checks_family_and_demo_oracle(family, oracle): schema["oracle"] = "path" if oracle != "path" else "sql" with pytest.raises(ValueError, match="not allowed"): entry.validate_spec(spec) + + +@pytest.mark.parametrize("mode,timeout", [("http", 90), ("cli", 30)]) +def test_verifier_deadline_uses_mode_budget_and_all_steps(tmp_path, monkeypatch, mode, timeout): + from openultrasast.search.budget import verification_run_seconds, verification_timeout + + schema = demo() + schema["start"]["mode"] = mode + schema["steps"] *= 8 + assert verification_timeout(schema) == timeout + assert verification_timeout(schema, {mode + "_start_seconds": 17}) == 17 + workloads = [] + + def lane(args, workload): + workloads.append(workload) + return lambda *args: asdict(record()) + + monkeypatch.setattr(transport, "AXLane", lane) + dispatch = transport.AXSideDispatcher( + SimpleNamespace(), "image@sha256:" + "a" * 64, tmp_path / "out", prepare=lambda side, spec: b"archive" + ) + dispatch(side=Side(tmp_path), demo=schema, family="injection", timeout_seconds=timeout, fresh_task=True) + assert workloads[0].deadline == verification_run_seconds(schema, timeout) + 60 == timeout * 9 + 65 diff --git a/tests/test_search_worker.py b/tests/test_search_worker.py index e17d55b..50534e2 100644 --- a/tests/test_search_worker.py +++ b/tests/test_search_worker.py @@ -327,7 +327,7 @@ def test_task_spend_finishes_with_confirmed_observations(tmp_path, with_evidence worker = Worker(model, repo=tmp_path, demo=tmp_path / "demo", prices=Prices(1, 1, 1), executor=InProcessExecutor(tmp_path)) job = task("explore") # First reservation fits; the second includes the real file contents and cannot fit. - job.limits["max_call_usd"] = 0.01 if with_evidence else 0.000001 + job.limits["max_call_usd"] = 0.014 if with_evidence else 0.000001 result = worker(job) assert result["end_detail"] == result["budget_exhaustion"]["end_detail"] == "task_spend" assert len(model.calls) == int(with_evidence) @@ -362,7 +362,7 @@ def test_calibrated_reservation_covers_reported_usage(tmp_path): expected_input = math.ceil(len(json.dumps([call["messages"], call["tools"]], ensure_ascii=False).encode()) / 2) + 1024 * ( len(call["messages"]) + 1 ) - assert worker.metrics["reserved"] == pytest.approx((expected_input + 1024 * 2) / 1_000_000) + assert worker.metrics["reserved"] == pytest.approx((expected_input + 4096 * 2) / 1_000_000) assert worker.metrics["reserved"] >= worker.metrics["settled"] == pytest.approx(0.029048) @@ -481,3 +481,36 @@ def test_plain_text_consumes_step(tmp_path): result = Worker(model, repo=tmp_path, demo=tmp_path / "demo", max_steps=1)(task("explore")) assert result["end_detail"] == "worker_steps" assert len(model.calls) == 1 + + +@pytest.mark.parametrize("finish_reason,expected_calls", [("length", 3), ("tool_calls", 2)]) +def test_truncated_arguments_retry_once_before_malformed(tmp_path, finish_reason, expected_calls): + broken = tool("run", command=["unused"]) + broken["tool_calls"][0]["function"]["arguments"] = '{"command":["unterminated' + + class LengthModel(StubModel): + def complete_chat_raw(self, **kwargs): + raw = super().complete_chat_raw(**kwargs) + raw["choices"][0]["finish_reason"] = finish_reason + return raw + + model = LengthModel([broken, broken, tool("finish", failure="shorter call worked")]) + worker = Worker(model, repo=tmp_path, demo=tmp_path / "demo") + result = worker(task("explore")) + assert len(model.calls) == expected_calls + assert all(call["extra_body"]["max_tokens"] == 4096 for call in model.calls) + if finish_reason == "length": + assert result["failure"] == "shorter call worked" + assert "shorter call" in model.calls[1]["messages"][-1]["content"] + else: + assert "two consecutive malformed" in result["failure"] + + +def test_default_forty_steps_and_ten_step_reminders(tmp_path): + model = StubModel([tool("list_files")] * 40) + worker = Worker(model, repo=tmp_path, demo=tmp_path / "demo", executor=InProcessExecutor(tmp_path)) + result = worker(task("explore")) + assert result["end_detail"] == "worker_steps" + assert len(model.calls) == SearchBudget().max_steps == 40 + for index in (10, 20, 30): + assert f"Remaining steps: {40 - index}" in model.calls[index]["messages"][-1]["content"]