search-with-proof: readiness from the budget with a probe, 4096-token tool turns, 40-step tasks

HTTP start waits up to 90 s (CLI 30 s) with TCP then HTTP probing; app stdout
and stderr tails are recorded; tool-call turns get 4096 output tokens with one
retry on truncation; tasks get 40 steps with reminders.

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

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

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"

tests/test_search_verify_transport.py

@@ -48,7 +48,9 @@ def test_dispatch_uses_fresh_tasks_and_only_data(tmp_path):
     assert result.isolation_mode == "task-boundary"
     assert len(result.runs) == 3
     assert len(calls) == 3 and calls[0].id != calls[1].id
-    assert all(i.id.startswith("search-verify-") for i in calls)
+    # The operator egress rules match the task-name prefix, which comes from the workload kind.
+    assert all(len(i.id) == 32 and i.id.isalnum() for i in calls)
+    assert len({i.id for i in calls}) == 3
 
 
 @pytest.mark.parametrize(
@@ -240,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

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"]