mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
Marius CiepluchandClaude Opus 5.5
e39a4882b0
First live run: the reason step got ~190 tokens and returned {} -> silent
no_evidence. Now: a bootstrap intent reads the candidate and its entry point; the
reason prompt states the goal, family oracles, demo schema and must propose
intents when none are open; empty replies retry once then end as reason_refused;
failure records keep messages.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
365 lines
14 KiB
Python
365 lines
14 KiB
Python
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from benchmarks.search.pilot_run import main
|
|
|
|
|
|
@pytest.mark.parametrize("side,outcome", [("vulnerable", "demonstrated"), ("fixed", "inconclusive")])
|
|
def test_probe_pilot_end_to_end(tmp_path, capsys, side, outcome):
|
|
output = tmp_path / "records.jsonl"
|
|
assert (
|
|
main(
|
|
[
|
|
"--dry-run",
|
|
"--side",
|
|
side,
|
|
"--pairs",
|
|
"0",
|
|
"--ceiling-usd",
|
|
"0.5",
|
|
"--out",
|
|
str(output),
|
|
"--private-root",
|
|
str(tmp_path / "private"),
|
|
]
|
|
)
|
|
== 0
|
|
)
|
|
records = [json.loads(line) for line in output.read_text().splitlines()]
|
|
search, summary = records
|
|
assert search["outcome"] == outcome
|
|
assert search["model_calls"] == (3 if side == "vulnerable" else 5)
|
|
assert search["input_bytes"] > 0
|
|
assert search["board_bytes"] > 0
|
|
assert search["spend"]["settled"] == 0
|
|
assert summary["searches"] == 1
|
|
assert summary["outcomes"] == {outcome: 1}
|
|
assert "app.py" not in output.read_text()
|
|
|
|
|
|
def test_ceiling_refuses_before_stub_call(tmp_path):
|
|
output = tmp_path / "records.jsonl"
|
|
main(
|
|
[
|
|
"--dry-run",
|
|
"--side",
|
|
"vulnerable",
|
|
"--pairs",
|
|
"0",
|
|
"--ceiling-usd",
|
|
"0",
|
|
"--out",
|
|
str(output),
|
|
"--private-root",
|
|
str(tmp_path / "private"),
|
|
]
|
|
)
|
|
record = json.loads(output.read_text().splitlines()[0])
|
|
assert record["model_calls"] == 0
|
|
assert record["end_reason"] == "budget_spent"
|
|
|
|
|
|
def make_pilot(tmp_path, *, side="fixed", dry_run=False):
|
|
from argparse import Namespace
|
|
|
|
from benchmarks.search.pilot_run import Pilot
|
|
from openultrasast.model.endpoint import Prices
|
|
from openultrasast.plane.budget import SpendBudget
|
|
from openultrasast.plane.memory import FileStore
|
|
from openultrasast.search.worker import StubModel
|
|
|
|
args = Namespace(side=side, dry_run=dry_run, model="scripted", max_tokens=128, verify_lane="vm", image="unused")
|
|
row = dict(
|
|
repo="https://github.com/private/project",
|
|
vulnerable="a" * 40,
|
|
fixed="b" * 40,
|
|
family="path",
|
|
language="python",
|
|
file="app.py",
|
|
function="main",
|
|
)
|
|
return Pilot(args, row, 0, tmp_path, FileStore(tmp_path / "board"), SpendBudget(1), StubModel([{"content": "{}"}]), Prices(1, 1, 1))
|
|
|
|
|
|
def test_fixed_verification_never_acquires_vulnerable_revision(tmp_path, monkeypatch):
|
|
from benchmarks.search import pilot_run
|
|
from openultrasast.search.probe import materialise
|
|
|
|
pilot = make_pilot(tmp_path)
|
|
sides, 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))
|
|
prepared = []
|
|
|
|
def prepare(revision, spec):
|
|
prepared.append(revision)
|
|
return b"executor-built-bundle"
|
|
|
|
monkeypatch.setattr(pilot, "prepare", prepare)
|
|
monkeypatch.setattr(pilot_run._sandbox, "isolation_mode", lambda: "userns")
|
|
|
|
# Inspect requested pins without relying on archive mechanics.
|
|
class StopVerification(Exception):
|
|
pass
|
|
|
|
def stop(archive, bundle):
|
|
bundle.mkdir()
|
|
(bundle / "spec.json").write_text(json.dumps({"demo": schema, "family": "path", "timeout_seconds": 5}))
|
|
|
|
monkeypatch.setattr(pilot_run, "extract_checkout", stop)
|
|
monkeypatch.setattr(pilot_run, "verify", lambda *args, **kwargs: (_ for _ in ()).throw(StopVerification()))
|
|
with pytest.raises(StopVerification):
|
|
pilot.verification()
|
|
assert prepared == ["b" * 40, "b" * 40]
|
|
assert "a" * 40 not in json.dumps(pilot.row)
|
|
|
|
|
|
def test_worker_failure_is_not_no_evidence(tmp_path):
|
|
pilot = make_pilot(tmp_path, dry_run=True)
|
|
from openultrasast.search.probe import materialise
|
|
|
|
pilot.sides, _ = materialise(tmp_path / "probe", "path")
|
|
pilot.worker.client.responses = iter([{"content": "invalid json"}, {"content": "{}"}, {"content": "{}"}])
|
|
record = pilot.run()
|
|
assert record["outcome"] == "inconclusive"
|
|
assert record["failures"][0]["phase"] == "explore"
|
|
assert "ValueError: explore must use tools or finish" in record["failures"][0]["reason"]
|
|
|
|
|
|
def test_shared_budget_reserved_and_settled_on_error(tmp_path):
|
|
from openultrasast.model.endpoint import Prices
|
|
from openultrasast.plane.budget import SpendBudget
|
|
from openultrasast.search.worker import Worker
|
|
|
|
run = SpendBudget(0.5)
|
|
task = SpendBudget(0.05)
|
|
|
|
class Client:
|
|
max_attempts = 1
|
|
|
|
def complete_chat_raw(self, **kwargs):
|
|
assert run.reserved == task.reserved > 0
|
|
raise RuntimeError("secret-provider-diagnostic")
|
|
|
|
worker = Worker(Client(), repo=tmp_path, demo=tmp_path / "demo", prices=Prices(1, 1, 1), run_budget=run)
|
|
with pytest.raises(RuntimeError):
|
|
worker.reason("facts: []", task)
|
|
assert run.reserved == task.reserved == 0
|
|
assert run.spent == task.spent == worker.metrics["settled"] > 0
|
|
assert worker.metrics["model_calls"] == 1
|
|
|
|
|
|
@pytest.mark.parametrize("verify_lane", ["vm", "ax"])
|
|
def test_live_wiring_with_fake_mailbox_no_network(tmp_path, monkeypatch, verify_lane):
|
|
import shutil
|
|
from dataclasses import asdict
|
|
|
|
from benchmarks.search import pilot_run
|
|
from openultrasast.search.demo import load_demo
|
|
from openultrasast.search.executor import InProcessExecutor
|
|
from openultrasast.search.probe import materialise
|
|
from openultrasast.search.verify import RunObservation, SideRecord
|
|
from openultrasast.search.verify_task import pack_checkout
|
|
|
|
pilot = make_pilot(tmp_path, side="vulnerable")
|
|
pilot.args.verify_lane = verify_lane
|
|
pilot.args.image = "registry.example/search@sha256:" + "0" * 64
|
|
sides, demos = materialise(tmp_path / "probe", "path")
|
|
pilot.worker.client = pilot_run.scripted(load_demo(demos["real"]))
|
|
events = []
|
|
active = []
|
|
|
|
class Mailbox:
|
|
def submit(self, name, args, **kwargs):
|
|
if name == "run":
|
|
assert args["command"][:3] == ["sh", "-ec", pilot_run.CLONE]
|
|
self.side = 0 if args["command"][-1] == "a" * 40 else 1
|
|
events.append(("clone", self.side))
|
|
return {"exit_code": 0}
|
|
return InProcessExecutor(sides[self.side].checkout).submit(name, args, **kwargs)
|
|
|
|
class Executor:
|
|
def __init__(self, lane, archive, *, deadline):
|
|
assert 0 < deadline <= 900
|
|
self.client = Mailbox()
|
|
|
|
def __enter__(self):
|
|
assert not active
|
|
active.append(self)
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
active.remove(self)
|
|
events.append(("cleanup", self.client.side))
|
|
|
|
def prepare(self, spec):
|
|
bundle = tmp_path / ("export-" + str(len(events)))
|
|
shutil.copytree(sides[self.client.side].checkout, bundle / "checkout")
|
|
(bundle / "products").mkdir()
|
|
(bundle / "spec.json").write_text(json.dumps(spec))
|
|
return pack_checkout(bundle)
|
|
|
|
def verify_lane_call(item, output, attempt):
|
|
assert not active # No executor consumes a slot during verification.
|
|
assert item.command == ("python3", "-m", "openultrasast.search.verify_task")
|
|
assert set(item.inputs) == {"VERIFY_INPUT_URL"}
|
|
events.append(("verify", None))
|
|
# The dispatcher's six fresh repetitions get owned observations.
|
|
observed = sum(kind == "verify" for kind, _ in events) <= 3
|
|
return asdict(SideRecord("observed", (RunObservation(observed, "owned", 0.1),), 0.1, "completed", isolation_mode="task-boundary"))
|
|
|
|
monkeypatch.setattr(pilot_run, "SearchExecutorTask", Executor)
|
|
pilot.lane = verify_lane_call
|
|
record = pilot.run()
|
|
assert record["outcome"] == "demonstrated", record
|
|
assert record["end_reason"] == "goal_met"
|
|
assert record["model_calls"] == 3
|
|
assert record["executor_tasks"] == 3
|
|
assert record["verify_tasks"] == (6 if verify_lane == "ax" else 0)
|
|
assert record["tasks_submitted"] == (9 if verify_lane == "ax" else 3)
|
|
assert record["input_bytes"] > 0
|
|
assert [event for event in events if event[0] == "clone"] == [("clone", 0), ("clone", 0), ("clone", 1)]
|
|
assert not active
|
|
assert "private/project" not in json.dumps(record)
|
|
calls = json.dumps(pilot.worker.client.calls)
|
|
assert "b" * 40 not in calls
|
|
assert "family" in calls and "main" in calls and "app.py" in calls
|
|
|
|
|
|
def test_no_oracle_does_not_acquire_or_build(tmp_path, monkeypatch):
|
|
from openultrasast.search.probe import materialise
|
|
|
|
pilot = make_pilot(tmp_path)
|
|
_, demos = materialise(tmp_path / "probe", "path")
|
|
(tmp_path / "demo/demo.json").write_bytes((demos["real"] / "demo.json").read_bytes())
|
|
pilot.family = "injection"
|
|
monkeypatch.setattr(pilot, "prepare", lambda *args: pytest.fail("no-oracle must not build"))
|
|
assert pilot.verification().outcome == "no_oracle"
|
|
assert pilot.record["tasks_submitted"] == 0
|
|
|
|
|
|
@pytest.mark.parametrize("ceiling", ["-1", "nan", "inf"])
|
|
def test_invalid_ceiling_rejected_before_work(tmp_path, ceiling):
|
|
with pytest.raises(SystemExit, match="2"):
|
|
main(["--dry-run", "--side", "fixed", "--pairs", "0", "--ceiling-usd", ceiling, "--out", str(tmp_path / "record")])
|
|
assert not (tmp_path / "record").exists()
|
|
|
|
|
|
@pytest.mark.parametrize("observed,expected", [([True] * 3, True), ([False] * 3, False), ([True, False, True], None)])
|
|
def test_fixed_effect_reporting_is_independent_of_differential(tmp_path, monkeypatch, observed, expected):
|
|
from dataclasses import asdict
|
|
|
|
from openultrasast.search.budget import SearchBudget
|
|
from openultrasast.search.coordinator import SearchTask
|
|
from openultrasast.search.verify import RunObservation, SideRecord, VerificationRecord
|
|
|
|
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)
|
|
task = SearchTask("task-1", "search-1", "verify", "", {}, asdict(SearchBudget()))
|
|
assert pilot.dispatch(task)["outcome"] == "inconclusive"
|
|
assert pilot.record["fixed_effect_observed"] is expected
|
|
|
|
|
|
def test_vm_preflight_refuses_before_client_or_store(tmp_path, monkeypatch):
|
|
from benchmarks.search import pilot_run
|
|
|
|
monkeypatch.setattr(pilot_run._sandbox, "isolation_mode", lambda: "task-boundary")
|
|
monkeypatch.setattr(pilot_run, "OpenRouterChatClient", lambda **kwargs: pytest.fail("client must not open"))
|
|
monkeypatch.setattr(pilot_run, "AXLane", lambda *args: pytest.fail("store must not open"))
|
|
with pytest.raises(SystemExit, match="2"):
|
|
main(
|
|
[
|
|
"--side",
|
|
"vulnerable",
|
|
"--pairs",
|
|
"0,1,2",
|
|
"--ceiling-usd",
|
|
"1.5",
|
|
"--image",
|
|
"registry.example/search@sha256:" + "0" * 64,
|
|
"--out",
|
|
str(tmp_path / "records"),
|
|
]
|
|
)
|
|
|
|
|
|
def test_actual_usage_settles_shared_budget_and_counts_tokens(tmp_path):
|
|
from openultrasast.model.endpoint import Prices
|
|
from openultrasast.plane.budget import SpendBudget
|
|
from openultrasast.search.worker import Worker
|
|
|
|
run, task = SpendBudget(0.5), SpendBudget(0.05)
|
|
|
|
class Client:
|
|
max_attempts = 1
|
|
|
|
def complete_chat_raw(self, **kwargs):
|
|
assert run.reserved == task.reserved > 0.0003
|
|
return {"choices": [{"message": {"content": "{}"}}], "usage": {"prompt_tokens": 100, "completion_tokens": 100}}
|
|
|
|
worker = Worker(Client(), repo=tmp_path, demo=tmp_path / "demo", prices=Prices(1, 1, 2), run_budget=run)
|
|
assert worker.reason("facts: []", task) == {}
|
|
assert worker.metrics["tokens_in"] == worker.metrics["tokens_out"] == 100
|
|
assert worker.metrics["settled"] == run.spent == task.spent == 0.0003
|
|
assert run.reserved == task.reserved == 0
|
|
|
|
|
|
def test_refused_shared_reservation_releases_task_allowance(tmp_path):
|
|
from openultrasast.model.endpoint import Prices
|
|
from openultrasast.plane.budget import BudgetExhausted, SpendBudget
|
|
from openultrasast.search.worker import StubModel, Worker
|
|
|
|
run, task = SpendBudget(0), SpendBudget(0.05)
|
|
client = StubModel([])
|
|
worker = Worker(client, repo=tmp_path, demo=tmp_path / "demo", prices=Prices(1, 1, 2), run_budget=run)
|
|
with pytest.raises(BudgetExhausted):
|
|
worker.reason("facts: []", task)
|
|
assert not client.calls
|
|
assert task.reserved == task.spent == run.reserved == run.spent == 0
|
|
|
|
|
|
@pytest.mark.parametrize("through_worker", [False, True])
|
|
def test_explore_exception_keeps_sanitized_message(tmp_path, monkeypatch, through_worker):
|
|
from contextlib import contextmanager
|
|
|
|
from openultrasast.search.worker import StubModel
|
|
|
|
pilot = make_pilot(tmp_path)
|
|
monkeypatch.setenv("MODEL_API_KEY", "private-token-1234")
|
|
|
|
def fail():
|
|
raise RuntimeError("mailbox failed: " + "app.py " * 100 + "https://host/private?token=secret private-token-1234: diagnostic tail")
|
|
|
|
class Broken(StubModel):
|
|
def complete_chat_raw(self, **kwargs):
|
|
fail()
|
|
|
|
@contextmanager
|
|
def executor(*args, **kwargs):
|
|
if not through_worker:
|
|
fail()
|
|
from types import SimpleNamespace
|
|
|
|
yield SimpleNamespace(client=None)
|
|
|
|
monkeypatch.setattr(pilot, "executor", executor)
|
|
if through_worker:
|
|
pilot.worker.client = Broken([])
|
|
record = pilot.run()
|
|
failure = record["failures"][0]
|
|
assert failure["phase"] == "explore"
|
|
assert failure["reason"].startswith("RuntimeError: ")
|
|
assert "diagnostic tail" in failure["reason"]
|
|
assert len(failure["reason"]) <= 300
|
|
assert "app.py" not in failure["reason"]
|
|
assert "https://" not in failure["reason"]
|
|
assert "private-token-1234" not in failure["reason"]
|