From 4b38239f687b3117125c2326fbcdf1ead67440b1 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Mon, 5 Oct 2026 14:26:43 +0200 Subject: [PATCH] search-with-proof group 4 + image: board, spend reservation, coordinator, search-task image Board as immutable versioned blobs (coordinator sole writer); budget reserve/settle before model calls; per-task and run caps; coordinator loop over an injected executor; batch.py accepts a fixed allow-list of search variables; XSS browser oracle; ousast-search-task image (Python, Node/TypeScript, PHP, Java 17, sqlite, bwrap, chromium) built by CI. 103 tests on the host. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/engine-image.yml | 3 + .kiro/specs/search-with-proof/tasks.md | 23 +- benchmarks/ax/batch.py | 21 +- .../measurements/2026-09-08-module-audit.json | 41 ++- plane/Dockerfile.search-task | 30 ++ src/openultrasast/model/audit.py | 3 + src/openultrasast/plane/budget.py | 68 +++- src/openultrasast/search/board.py | 114 ++++++ src/openultrasast/search/budget.py | 37 ++ src/openultrasast/search/coordinator.py | 348 ++++++++++++++++++ src/openultrasast/search/oracles.py | 103 +++++- tests/test_ax_batch.py | 18 + tests/test_search_board.py | 39 ++ tests/test_search_browser.py | 50 +++ tests/test_search_budget.py | 58 +++ tests/test_search_coordinator.py | 166 +++++++++ tests/test_search_image.py | 18 + 17 files changed, 1120 insertions(+), 20 deletions(-) create mode 100644 plane/Dockerfile.search-task create mode 100644 src/openultrasast/search/board.py create mode 100644 src/openultrasast/search/budget.py create mode 100644 src/openultrasast/search/coordinator.py create mode 100644 tests/test_search_board.py create mode 100644 tests/test_search_browser.py create mode 100644 tests/test_search_budget.py create mode 100644 tests/test_search_coordinator.py create mode 100644 tests/test_search_image.py diff --git a/.github/workflows/engine-image.yml b/.github/workflows/engine-image.yml index 764f092..078dc75 100644 --- a/.github/workflows/engine-image.yml +++ b/.github/workflows/engine-image.yml @@ -11,6 +11,7 @@ on: - 'benchmarks/learn/engine_worker_service.py' - 'Dockerfile' - 'plane/Dockerfile.engine-task' + - 'plane/Dockerfile.search-task' - 'ops/frontend-retention/install.py' - 'pyproject.toml' - '.github/workflows/engine-image.yml' @@ -35,6 +36,8 @@ jobs: - package: ousast-replay-task dockerfile: plane/Dockerfile.engine-task target: replay + - package: ousast-search-task + dockerfile: plane/Dockerfile.search-task steps: - uses: actions/checkout@v5 - uses: docker/setup-buildx-action@v3 diff --git a/.kiro/specs/search-with-proof/tasks.md b/.kiro/specs/search-with-proof/tasks.md index 71d2ce0..f11cad0 100644 --- a/.kiro/specs/search-with-proof/tasks.md +++ b/.kiro/specs/search-with-proof/tasks.md @@ -29,13 +29,13 @@ gates, runs every networked or paid step. are unreachable. Record `benchmarks/measurements/2026-10-xx-search-probe/record.json`. _Requirements: 6.2_ - [ ] 3.3 Gate: every oracle works under gVisor, or the family is marked `no_oracle` for the pilot. -- [ ] 4. Board, coordinator, budgets - - [ ] 4.1 Board in the memory store as immutable versioned blobs (`plane/memory.py`); facts, intents, hints with +- [x] 4. Board, coordinator, budgets + - [x] 4.1 Board in the memory store as immutable versioned blobs (`plane/memory.py`); facts, intents, hints with writer step and task; coordinator is the only writer; tasks read and write through presigned links. _Requirements: 1.1-1.3_ - - [ ] 4.2 Spend reservation before each model call in `plane/budget.py`, settled after; per-task memory, disk, + - [x] 4.2 Spend reservation before each model call in `plane/budget.py`, settled after; per-task memory, disk, cleanup caps; total wall-time and retry caps; tests that in-flight calls cannot overrun. _Requirements: 4.1, 4.2_ - - [ ] 4.3 Dispatch reason, explore and verify as AX Tasks through `benchmarks/ax/batch.py` (extend its environment + - [x] 4.3 Dispatch reason, explore and verify as AX Tasks through `benchmarks/ax/batch.py` (extend its environment contract beyond numeric deadlines), VM lane as fallback, resume from checkpoints. _Requirements: 2.3, 2.4_ - [ ] 5. Agent worker @@ -65,3 +65,18 @@ gates, runs every networked or paid step. without isolation. Artefacts receive a batch CLI `TARGET` bridge and private scratch; HTTP apps need a trusted CLI adapter. Only trusted `Side.build_command` outputs can enter the app runtime (read-only `/build`); artefact `build.sh` writes scratch only. XSS requires a trusted browser executor, otherwise `no_oracle`. + +- Tasks 3.1 (files only) and group 4 (2026-10-05): independent review approved; final offline validation: + 107 passed, 2 skipped (Chromium namespace/socket restrictions and the existing SSRF localhost restriction). + Ruff and changed-source mypy pass; module audit: 160 load-bearing, 6 standalone, zero orphaned. No network + actions, model calls, image builds, pushes or commits were performed. Task 3.1 remains unchecked until the + image is built/published and a real digest is available; CI now includes `plane/Dockerfile.search-task`. +- The coordinator accepts an injected executor. `BatchExecutor` uses the repository's `benchmarks.ax.batch` + transport with one attempt per admission; coordinator owns the global retry budget and VM fallback. It requires + a repository checkout import path and a trusted worker command; reason/explore workers remain group 5. + Per-task memory, disk and cleanup ceilings are passed in the worker contract; worker enforcement and gVisor + compatibility still require group 5 and the 3.2 probe. Only trusted verification results can meet the goal. +- Board heads are immutable memory-store hashes; retain the returned head with the host run to resume. Unknown + interrupted model calls conservatively consume their full reservation; settled failures with zero spend release + it. Cleanup exceptions stop dispatch. Chromium uses a fresh profile in a private network namespace; injected + script must call `alert("ousast-xss")`, while escaped/reflected payloads cannot produce the verifier DOM marker. diff --git a/benchmarks/ax/batch.py b/benchmarks/ax/batch.py index b5feec1..990fa50 100644 --- a/benchmarks/ax/batch.py +++ b/benchmarks/ax/batch.py @@ -280,11 +280,18 @@ def manifest(name, workload, item, urls, atespace="default", *, validate_name=Tr raise ValueError("invalid AX task name") if set(urls) != workload.url_env: raise ValueError("unexpected URL environment names") - # Extra env is deliberately a small numeric execution contract, never inherited credentials. - if set(item.extra_env) - {"DEADLINE", "QUESTION_DEADLINE"} or any( - not re.fullmatch(r"[0-9]+(?:\.[0-9]+)?", str(v)) for v in item.extra_env.values() - ): + # Fixed names and semantic values, never inherited environment or credentials. + patterns = { + "DEADLINE": r"[0-9]+(?:\.[0-9]+)?", + "QUESTION_DEADLINE": r"[0-9]+(?:\.[0-9]+)?", + "SEARCH_STEP": r"reason|explore|verify", + "SEARCH_ID": r"[a-z0-9][a-z0-9-]{0,63}", + "SEARCH_TASK_ID": r"[a-z0-9][a-z0-9-]{0,63}", + } + if any(key not in patterns or not re.fullmatch(patterns[key], str(value)) for key, value in item.extra_env.items()): raise ValueError("unexpected task environment") + if set(item.extra_env) & set(urls): + raise ValueError("overlapping task environment") for url in urls.values(): public_endpoint(url) env = [{"name": key, "value": str(value)} for key, value in {**urls, **item.extra_env}.items()] @@ -358,8 +365,10 @@ def schedule(pending, lanes, out, execute, complete, *, fallbacks=(), overflow=T return failures -def dispatch(items, workload, out, executors, *, lanes=("ax", "ax", "docker"), overflow=True): +def dispatch(items, workload, out, executors, *, lanes=("ax", "ax", "docker"), overflow=True, max_attempts=2): """Digest checkpoints, one instrument rerun, sandbox retry, priority VM fallback.""" + if not isinstance(max_attempts, int) or not 1 <= max_attempts <= 2: + raise ValueError("batch attempts must be one or two") out.mkdir(parents=True, exist_ok=True) pending, fallbacks = [], [] @@ -380,7 +389,7 @@ def dispatch(items, workload, out, executors, *, lanes=("ax", "ax", "docker"), o output = out / hashlib.sha256(item.id.encode()).hexdigest() output.mkdir(exist_ok=True) attempts = [] - for attempt in range(2): + for attempt in range(max_attempts): attempt_output = output / ("attempt-" + uuid.uuid4().hex) attempt_output.mkdir() record = executors[lane](item, attempt_output, attempt) diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json index 767e64a..7c7da85 100644 --- a/benchmarks/measurements/2026-09-08-module-audit.json +++ b/benchmarks/measurements/2026-09-08-module-audit.json @@ -1,8 +1,8 @@ { - "modules": 163, + "modules": 166, "counts": { - "load_bearing": 158, - "standalone_capability": 5 + "load_bearing": 160, + "standalone_capability": 6 }, "orphaned": [], "standalone": { @@ -10,6 +10,7 @@ "mcp": "narrow MCP server over stdio; its own `ousast mcp` entry point and OpenCode integration", "push_inputs": "pre-push-safety-net task 1.2: offline input provenance validator; python -m openultrasast.push_inputs", "push_scoring": "pre-push-safety-net task 1.3: frozen-profile diagnostic outcome scorer; python -m openultrasast.push_scoring", + "search.coordinator": "search-with-proof Req 1/2/4: injected search executor and offline board checkpoint inspector; agent worker pending", "skills": "skill router; consumed by mcp and by the OpenCode integration, not by the scan path" }, "rows": [ @@ -897,7 +898,9 @@ "plane.tasks.remember", "plane.tasks.roles", "plane.tasks.verify", - "plane.workspaces" + "plane.workspaces", + "search.board", + "search.coordinator" ] }, { @@ -913,7 +916,8 @@ "learn.program", "plane.generate", "plane.tasks.roles", - "plane.tasks.verify" + "plane.tasks.verify", + "search.coordinator" ] }, { @@ -1000,7 +1004,9 @@ "plane.generate", "plane.reconciler", "plane.tasks.loop", - "plane.tasks.remember" + "plane.tasks.remember", + "search.board", + "search.coordinator" ] }, { @@ -1599,6 +1605,7 @@ "classification": "load_bearing", "reason": "reachable from gate, map_gate, pair_gate, cli", "importers": [ + "search.coordinator", "search.verify" ] }, @@ -1610,6 +1617,28 @@ "search.verify" ] }, + { + "module": "search.board", + "classification": "load_bearing", + "reason": "reachable from gate, map_gate, pair_gate, cli", + "importers": [ + "search.coordinator" + ] + }, + { + "module": "search.budget", + "classification": "load_bearing", + "reason": "reachable from gate, map_gate, pair_gate, cli", + "importers": [ + "search.coordinator" + ] + }, + { + "module": "search.coordinator", + "classification": "standalone_capability", + "reason": "search-with-proof Req 1/2/4: injected search executor and offline board checkpoint inspector; agent worker pending", + "importers": [] + }, { "module": "search.oracles", "classification": "load_bearing", diff --git a/plane/Dockerfile.search-task b/plane/Dockerfile.search-task new file mode 100644 index 0000000..972a381 --- /dev/null +++ b/plane/Dockerfile.search-task @@ -0,0 +1,30 @@ +# Build from the repository root; retain the AX runner entrypoint and command. +FROM eclipse-temurin:17-jdk-jammy AS jdk +FROM ghcr.io/norandom/ax-task-runner:v0.3.1 +LABEL org.opencontainers.image.licenses="Apache-2.0" \ + org.opencontainers.image.source="https://github.com/norandom/OpenUltraSAST" +USER root +COPY --from=jdk /opt/java/openjdk /opt/java/openjdk +ENV JAVA_HOME=/opt/java/openjdk +ENV PATH="/venv/bin:/opt/java/openjdk/bin:${PATH}" +# dpkg exclusions prevent docs/manpages entering layers, including transitive deps. +RUN printf 'path-exclude=/usr/share/doc/*\npath-exclude=/usr/share/man/*\npath-exclude=/usr/share/info/*\n' \ + > /etc/dpkg/dpkg.cfg.d/01-no-docs \ + && apt-get update \ + && apt-get install -y --no-install-recommends \ + python3 python3-venv nodejs node-typescript php-cli sqlite3 bubblewrap util-linux chromium git ca-certificates \ + && rm -rf /var/lib/apt/lists/* /opt/java/openjdk/lib/src.zip /opt/java/openjdk/man +WORKDIR /app +COPY pyproject.toml README.md LICENSE ./ +COPY src ./src +# Includes the verifier-owned oracle fixtures and sandbox helpers. The worker +# protocol is implemented separately (search-with-proof group 5). +RUN python3 -m venv /venv \ + && /venv/bin/pip install --no-cache-dir . \ + && python3 --version && node --version && tsc --version && php --version \ + && javac -version && sqlite3 --version && bwrap --version \ + && prlimit --version && nsenter --version && chromium --version +# Browser sandbox requires a non-root caller; the AX runner needs no root privileges. +RUN useradd --create-home --uid 10001 search +USER search +# Do not replace the AX runner's ENTRYPOINT or CMD. diff --git a/src/openultrasast/model/audit.py b/src/openultrasast/model/audit.py index 77678e1..ba44ae6 100644 --- a/src/openultrasast/model/audit.py +++ b/src/openultrasast/model/audit.py @@ -83,6 +83,9 @@ SPEC_OWNED = { # Reachable modules that stand alone: their own CLI subcommand or external protocol, off the scan path. STANDALONE = { + "search.coordinator": ( + "search-with-proof Req 1/2/4: injected search executor and offline board checkpoint inspector; agent worker pending" + ), "push_scoring": "pre-push-safety-net task 1.3: frozen-profile diagnostic outcome scorer; python -m openultrasast.push_scoring", "push_inputs": "pre-push-safety-net task 1.2: offline input provenance validator; python -m openultrasast.push_inputs", "mcp": "narrow MCP server over stdio; its own `ousast mcp` entry point and OpenCode integration", diff --git a/src/openultrasast/plane/budget.py b/src/openultrasast/plane/budget.py index 4f24d78..df76415 100644 --- a/src/openultrasast/plane/budget.py +++ b/src/openultrasast/plane/budget.py @@ -10,6 +10,7 @@ failure is `AccountError` (the task is `failed` and the run starts nothing furth from __future__ import annotations from collections.abc import Mapping +from decimal import Decimal from typing import Any, Protocol from ..model.endpoint import Prices @@ -251,5 +252,70 @@ class MeteredEmbeddingClient: __all__ = [ "AccountError", "BudgetExhausted", "EmbeddingClient", "MeteredClient", "MeteredEmbeddingClient", "UnmeteredUsage", "cost_of", - "is_account_error", "prices_from", + "is_account_error", "prices_from", "SpendBudget", "SpendReservation", ] # fmt: skip + + +class SpendBudget: + """Thread-safe pre-call admission. Share one instance across concurrent callers. + + A caller must supply a genuine upper bound, including all tool-loop calls. + Reservations remain charged until settled or explicitly released on failure. + Existing MeteredClient callers retain their historical post-call metering. + """ + + def __init__(self, cap: float, *, spent: float = 0.0) -> None: + from threading import Lock + + self._lock = Lock() + self._cap = self._amount(cap) + self._spent = self._amount(spent) + if self._spent > self._cap: + raise ValueError("spend exceeds cap") + self._reserved = Decimal(0) + + @staticmethod + def _amount(value: float) -> Decimal: + + amount = Decimal(str(value)) + if not amount.is_finite() or amount < 0: + raise ValueError("cost must be finite and nonnegative") + return amount + + @property + def spent(self) -> float: + with self._lock: + return float(self._spent) + + @property + def reserved(self) -> float: + with self._lock: + return float(self._reserved) + + def reserve(self, max_cost: float) -> SpendReservation: + amount = self._amount(max_cost) + with self._lock: + if self._spent + self._reserved + amount > self._cap: + raise BudgetExhausted("spend reservation exceeds cap", usd=float(self._spent), calls=0) + self._reserved += amount + return SpendReservation(self, amount) + + +class SpendReservation: + def __init__(self, budget: SpendBudget, amount: Decimal) -> None: + self._budget, self._amount, self._active = budget, amount, True + + def settle(self, actual: float) -> None: + amount = self._budget._amount(actual) + with self._budget._lock: + if not self._active: + raise ValueError("reservation already closed") + if amount > self._amount: + # Retain the reservation: a broken upper bound must fail closed. + raise ValueError("actual cost exceeds reservation") + self._budget._reserved -= self._amount + self._budget._spent += amount + self._active = False + + def release(self) -> None: + self.settle(0) diff --git a/src/openultrasast/search/board.py b/src/openultrasast/search/board.py new file mode 100644 index 0000000..f3a7da4 --- /dev/null +++ b/src/openultrasast/search/board.py @@ -0,0 +1,114 @@ +"""Coordinator-owned, content-addressed search checkpoints in the plane memory store. + +Workers receive snapshots and return proposals; they never receive this writer or +store credentials. Persist the returned head in the host's run checkpoint. +""" + +from __future__ import annotations + +import copy +import hashlib +import json +import re +import threading +from typing import Any + +import yaml + +from ..plane.memory import MemoryStore + + +class Board: + def __init__(self, store: MemoryStore, search_id: str, *, coordinator: str, max_bytes: int = 262144) -> None: + if not re.fullmatch(r"[a-z0-9][a-z0-9-]{0,63}", search_id) or not coordinator or max_bytes < 256: + raise ValueError("invalid board configuration") + self.store, self.coordinator, self.max_bytes = store, coordinator, max_bytes + self.prefix = "boards/" + hashlib.sha256(search_id.encode()).hexdigest() + self._lock = threading.RLock() + self.head: str | None = None + self._state: dict[str, Any] = dict( + search_id=search_id, version=0, previous=None, facts=[], intents=[], hints=[], end_reason=None, checkpoint={} + ) + + @property + def state(self) -> dict[str, Any]: + with self._lock: + return copy.deepcopy(self._state) + + @classmethod + def resume(cls, store: MemoryStore, search_id: str, head: str, *, coordinator: str, max_bytes: int = 262144) -> Board: + board = cls(store, search_id, coordinator=coordinator, max_bytes=max_bytes) + data = store.get_blob(board.prefix, head) + if data is None or len(data) > max_bytes: + raise ValueError("missing or oversized board checkpoint") + state = json.loads(data) + if state["search_id"] != search_id: + raise ValueError("checkpoint belongs to another search") + board._validate(state) + board._state, board.head = state, head + return board + + def commit(self, *, writer: str, **changes: Any) -> str: + with self._lock: + if writer != self.coordinator: + raise PermissionError("only the coordinator may write the board") + if self._state["end_reason"] is not None: + raise ValueError("search already ended") + if changes.keys() - {"facts", "intents", "hints", "end_reason", "checkpoint"}: + raise ValueError("unknown board fields") + state = {**self.state, **copy.deepcopy(changes), "version": self._state["version"] + 1, "previous": self.head} + self._validate(state) + data = json.dumps(state, sort_keys=True, allow_nan=False).encode() + if len(data) > self.max_bytes: + raise ValueError("board size cap exceeded") + head = self.store.put_blob(self.prefix, data) + self._state, self.head = state, head + return head + + @staticmethod + def _validate(state: dict[str, Any]) -> None: + if state["end_reason"] not in (None, "goal_met", "budget_spent", "exhausted"): + raise ValueError("invalid end reason") + ids: dict[str, set[str]] = {} + for kind in ("facts", "intents", "hints"): + ids[kind] = set() + for row in state[kind]: + required = ("id", "step", "task_id", "description" if kind == "intents" else "text") + if any(not isinstance(row.get(key), str) or not row[key] for key in required): + raise ValueError("entry requires text and writer provenance") + if row["id"] in ids[kind]: + raise ValueError("duplicate board id") + ids[kind].add(row["id"]) + for fact in state["facts"]: + if not isinstance(fact.get("evidence_refs"), list) or not fact["evidence_refs"]: + raise ValueError("facts require evidence references") + if any(not isinstance(ref, str) or not ref for ref in fact["evidence_refs"]): + raise ValueError("invalid evidence reference") + if not isinstance(fact.get("from_intents"), list) or set(fact["from_intents"]) - ids["intents"]: + raise ValueError("unknown source intent") + for intent in state["intents"]: + if not isinstance(intent.get("from_facts"), list) or set(intent["from_facts"]) - ids["facts"]: + raise ValueError("unknown source fact") + if intent.get("status") not in ("open", "claimed", "concluded"): + raise ValueError("invalid intent status") + if intent["status"] == "claimed" and not intent.get("claimant"): + raise ValueError("claimed intent requires claimant") + if intent["status"] == "open" and intent.get("claimant") is not None: + raise ValueError("open intent cannot have claimant") + + def snapshot(self, *, max_bytes: int = 32768) -> str: + """Valid YAML even when truncated; explicit omission counts, never a cut string.""" + if max_bytes < 128: + raise ValueError("snapshot cap must be at least 128 bytes") + state = self.state + view = {key: state[key] for key in ("version", "end_reason", "facts", "intents", "hints")} + view["omitted"] = dict.fromkeys(("facts", "intents", "hints"), 0) + while True: + text = str(yaml.safe_dump(view, sort_keys=False, allow_unicode=True)) + if len(text.encode()) <= max_bytes: + return text + kind = next((k for k in ("facts", "intents", "hints") if view[k]), None) + if kind is None: + raise ValueError("snapshot cap too small") + view[kind].pop(0) + view["omitted"][kind] += 1 diff --git a/src/openultrasast/search/budget.py b/src/openultrasast/search/budget.py new file mode 100644 index 0000000..8482fef --- /dev/null +++ b/src/openultrasast/search/budget.py @@ -0,0 +1,37 @@ +"""Search ceilings carried to every executor; no model or transport dependencies.""" + +import math +from dataclasses import dataclass + + +@dataclass(frozen=True) +class SearchBudget: + spend_usd: float = 0.50 + max_call_usd: float = 0.05 + reason_rounds: int = 4 + intents_per_round: int = 3 + demonstrations: int = 3 + tasks: int = 32 + retries: int = 1 + memory_bytes: int = 5 * 1024**3 + disk_bytes: int = 2 * 1024**3 + task_wall_seconds: float = 900 + wall_seconds: float = 4 * 3600 + cleanup_seconds: float = 30 + + def __post_init__(self) -> None: + for name, value in vars(self).items(): + if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value): + raise ValueError(f"invalid {name}") + if value < 0 or (value == 0 and name not in ("retries", "spend_usd")): + raise ValueError(f"invalid {name}") + if name in ( + "reason_rounds", + "intents_per_round", + "demonstrations", + "tasks", + "retries", + "memory_bytes", + "disk_bytes", + ) and not isinstance(value, int): + raise ValueError(f"{name} must be an integer") diff --git a/src/openultrasast/search/coordinator.py b/src/openultrasast/search/coordinator.py new file mode 100644 index 0000000..1dcb55d --- /dev/null +++ b/src/openultrasast/search/coordinator.py @@ -0,0 +1,348 @@ +"""Host-only board driver. Executors own task isolation and enforce supplied limits. + +Reason/explore workers are group 5. They return proposals, never board writes. +Only the trusted verify executor may return a demonstrated outcome. An interrupted +call is conservatively charged its reservation on resume, never silently free. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import time +from collections.abc import Callable +from dataclasses import asdict, dataclass, replace +from importlib import import_module +from pathlib import Path +from typing import Any, Protocol + +from ..plane.budget import BudgetExhausted, SpendBudget +from ..plane.memory import open_store +from .board import Board +from .budget import SearchBudget + + +@dataclass(frozen=True) +class SearchTask: + id: str + search_id: str + step: str + snapshot: str + payload: dict[str, Any] + limits: dict[str, Any] + lane: str = "ax" + + def __post_init__(self) -> None: + for value in (self.id, self.search_id): + if not re.fullmatch(r"[a-z0-9][a-z0-9-]{0,63}", value): + raise ValueError("invalid search task identifier") + if self.step not in ("reason", "explore", "verify") or self.lane not in ("ax", "kind", "docker"): + raise ValueError("invalid search task step or lane") + + +class Executor(Protocol): + def __call__(self, task: SearchTask) -> dict[str, Any]: + """Finish/clean up within limits; report actual cost, including failed calls. + + A missing cost on a model task charges its entire reservation. Model + workers must reserve each call against limits.max_call_usd (the task's + total allowance) using plane.budget.SpendBudget. Exceptions are untrusted + diagnostics and must never be copied into the board. + """ + ... + + +class Coordinator: + def __init__( + self, + board: Board, + executor: Executor, + *, + budget: SearchBudget | None = None, + clock: Callable[[], float] = time.time, + push_budget: SpendBudget | None = None, + ) -> None: + self.board, self.executor = board, executor + self.budget, self.clock, self.push_budget = budget or SearchBudget(), clock, push_budget + self.state = board.state + self.progress = self.state["checkpoint"] or dict( + started_at=clock(), + spent_usd=0.0, + tasks=0, + rounds=0, + retries=0, + demonstrations=0, + pending=None, + verified=[], + failures=[], + ) + self.meter = SpendBudget(self.budget.spend_usd, spent=self.progress["spent_usd"]) + + def _save(self) -> None: + self.progress["wall_seconds"] = max(0, self.clock() - self.progress["started_at"]) + self.state["checkpoint"] = self.progress + self.board.commit( + writer=self.board.coordinator, **{k: self.state[k] for k in ("facts", "intents", "hints", "end_reason", "checkpoint")} + ) + + def _end(self, reason: str) -> dict[str, Any]: + self.state["end_reason"] = reason + self._save() + return self.board.state + + def _remaining(self) -> float: + return float(self.budget.wall_seconds - max(0, self.clock() - self.progress["started_at"])) + + def _call(self, step: str, payload: dict[str, Any]) -> dict[str, Any]: + lane = "ax" + while True: + remaining = self._remaining() + if remaining <= self.budget.cleanup_seconds or self.progress["tasks"] >= self.budget.tasks: + raise BudgetExhausted("task or wall budget exhausted", usd=self.meter.spent, calls=self.progress["tasks"]) + maximum = self.budget.max_call_usd if step != "verify" else 0 + reservation = self.meter.reserve(maximum) + push_reservation = None + try: + if self.push_budget is not None: + push_reservation = self.push_budget.reserve(maximum) + except BudgetExhausted: + reservation.release() + raise + task_id = "task-" + str(self.progress["tasks"] + 1) + limits = asdict(self.budget) + limits["task_wall_seconds"] = min(self.budget.task_wall_seconds, remaining - self.budget.cleanup_seconds) + task = SearchTask(task_id, self.state["search_id"], step, self.board.snapshot(), payload, limits, lane) + self.progress["tasks"] += 1 + self.progress["pending"] = {"id": task_id, "step": step, "payload": payload} + # A crash after this durable write may have spent the whole allowance. + self.progress["spent_usd"] = self.meter.spent + maximum + if step == "explore": + intent = next(i for i in self.state["intents"] if i["id"] == payload["intent_id"]) + intent.update(status="claimed", claimant=task_id) + self._save() + task = replace(task, snapshot=self.board.snapshot()) + dispatched_at = self.clock() + try: + result = self.executor(task) + except Exception: + # Includes cleanup failures: never launch more work after a task + # may have leaked. The persisted reservation remains conservative. + reservation.settle(maximum) + if push_reservation is not None: + push_reservation.settle(maximum) + raise + if not isinstance(result, dict): + result = {"status": "invalid_result"} + actual = result.get("cost_usd", maximum) + try: + reservation.settle(actual) + if push_reservation is not None: + push_reservation.settle(actual) + except (ValueError, TypeError, ArithmeticError): + # Charge the maximum and stop on a violated executor contract. + if reservation._active: + reservation.settle(maximum) + if push_reservation is not None and push_reservation._active: + push_reservation.settle(maximum) + result = {"status": "invalid_cost"} + self.progress["spent_usd"] = self.meter.spent + self.progress["pending"] = None + if ( + result.get("status") == "invalid_cost" + or self._remaining() <= 0 + or self.clock() - dispatched_at > limits["task_wall_seconds"] + ): + raise BudgetExhausted("executor exceeded cost or wall contract", usd=self.meter.spent, calls=self.progress["tasks"]) + result = {**result, "task_id": task_id} + # Persist results together with their effects in run(), not a "done" + # checkpoint which could lose the result if the host then crashes. + if result.get("status") in ("sandbox_failure", "oom", "instrument_failure") and self.progress["retries"] < self.budget.retries: + self.progress["retries"] += 1 + lane = "docker" + continue + return result + + def run(self) -> dict[str, Any]: + if self.state["end_reason"]: + return self.board.state + pending = self.progress.get("pending") + if pending: + # Do not duplicate a possibly paid operation. Recover the claim as a + # recorded failure; the next reason task can propose another direction. + self.progress["failures"].append({"task_id": pending["id"], "reason": "interrupted"}) + for intent in self.state["intents"]: + if intent.get("claimant") == pending["id"]: + intent.update(status="concluded", failure="interrupted") + if pending["step"] == "verify": + self.progress["verified"].append(pending["payload"]["fact_id"]) + self.progress["pending"] = None + self._save() + while True: + if self._remaining() <= self.budget.cleanup_seconds: + return self._end("budget_spent") + try: + intent = next((i for i in self.state["intents"] if i["status"] == "open"), None) + demo = next((f for f in self.state["facts"] if f.get("demo") and f["id"] not in self.progress["verified"]), None) + if demo is not None: + if self.progress["demonstrations"] >= self.budget.demonstrations: + return self._end("budget_spent") + self.progress["demonstrations"] += 1 + result = self._call("verify", {"fact_id": demo["id"], "demo": demo["demo"]}) + self.progress["verified"].append(demo["id"]) + refs = result.get("evidence_refs") + if result.get("status") == "ok" and result.get("outcome") == "demonstrated" and self._refs(refs): + self.state["facts"].append( + dict( + id="verified-" + result["task_id"], + text="Differential demonstrated", + evidence_refs=refs, + from_intents=demo["from_intents"], + step="verify", + task_id=result["task_id"], + ) + ) + return self._end("goal_met") + self.progress["failures"].append({"task_id": result["task_id"], "reason": "not_demonstrated"}) + elif intent is not None: + result = self._call("explore", {"intent_id": intent["id"], "description": intent["description"]}) + intent["status"] = "concluded" + fact = result.get("fact") + if ( + result.get("status") == "ok" + and isinstance(fact, dict) + and isinstance(fact.get("text"), str) + and fact["text"] + and self._refs(fact.get("evidence_refs")) + ): + entry = dict( + id="fact-" + result["task_id"], + text=fact["text"], + evidence_refs=fact["evidence_refs"], + from_intents=[intent["id"]], + step="explore", + task_id=result["task_id"], + ) + if isinstance(fact.get("demo"), str) and fact["demo"]: + entry["demo"] = fact["demo"] + self.state["facts"].append(entry) + else: + intent["failure"] = "invalid_result" if result.get("status") == "ok" else result.get("status", "execution_failure") + else: + if self.progress["rounds"] >= self.budget.reason_rounds: + return self._end("budget_spent") + self.progress["rounds"] += 1 + result = self._call("reason", {}) + proposals = result.get("intents", []) + if result.get("status") != "ok" or not isinstance(proposals, list) or len(proposals) > self.budget.intents_per_round: + return self._end("exhausted") + known = {f["id"] for f in self.state["facts"]} + descriptions = {i["description"] for i in self.state["intents"]} + added = 0 + for proposal in proposals: + if not isinstance(proposal, dict): + continue + description, refs = proposal.get("description"), proposal.get("from_facts", []) + if not isinstance(description, str) or not description.strip() or description in descriptions: + continue + if not isinstance(refs, list) or any(not isinstance(ref, str) or ref not in known for ref in refs): + continue + added += 1 + descriptions.add(description) + self.state["intents"].append( + dict( + id=result["task_id"] + "-intent-" + str(added), + description=description, + from_facts=refs, + status="open", + claimant=None, + step="reason", + task_id=result["task_id"], + ) + ) + if not added: + return self._end("exhausted") + self._save() + except BudgetExhausted: + return self._end("budget_spent") + + @staticmethod + def _refs(refs: Any) -> bool: + return isinstance(refs, list) and bool(refs) and all(isinstance(ref, str) and bool(ref) for ref in refs) + + +class BatchExecutor: + """Adapter to the existing AX/VM transfer and checkpoint protocol. + + Pass AXLane/DockerLane instances and a trusted worker command. No default + worker is invented here: reason/explore entrypoints belong to group 5. + The coordinator owns retries, so batch performs exactly one attempt. + """ + + def __init__(self, *, image: str, command: tuple[str, ...], output: Path, lanes: dict[str, Any]) -> None: + batch = import_module("benchmarks.ax.batch") + batch.validate_image(image) + if not command: + raise ValueError("worker command required") + self.image, self.command, self.output, self.lanes = image, command, output, lanes + + def __call__(self, task: SearchTask) -> dict[str, Any]: + import copy + + batch = import_module("benchmarks.ax.batch") + + if task.lane not in self.lanes: + return {"status": "sandbox_failure", "cost_usd": 0} + data = json.dumps(asdict(task), sort_keys=True, allow_nan=False).encode() + digest = hashlib.sha256(data).hexdigest() + identity = task.search_id + "-" + task.id + item = batch.Item( + identity, + {"SPEC_URL": data}, + "result.dat", + self.command, + {"SEARCH_STEP": task.step, "SEARCH_ID": task.search_id, "SEARCH_TASK_ID": task.id}, + digest, + ) + + def validate(raw: bytes, output: Path) -> None: + if len(raw) > 262144: + raise ValueError("search result size exceeded") + record = json.loads(raw) + if not isinstance(record, dict) or record.get("status") not in ( + "ok", + "sandbox_failure", + "instrument_failure", + "oom", + "execution_failure", + ): + raise ValueError("invalid search result") + (output / "result.json").write_text(json.dumps(record)) + + workload = batch.Workload(self.image, frozenset({"SPEC_URL", "RESULT_URL"}), task.limits["task_wall_seconds"], validate) + lane = copy.copy(self.lanes[task.lane]) + if hasattr(lane, "workload"): + lane.workload = workload + out = self.output / task.search_id / task.id + batch.dispatch([item], workload, out, {task.lane: lane}, lanes=(task.lane,), max_attempts=1) + record_path = out / (hashlib.sha256(identity.encode()).hexdigest() + ".json") + if not record_path.exists(): + return {"status": "execution_failure", "cost_usd": 0} + record: dict[str, Any] = json.loads(record_path.read_text()) + return record + + +def main() -> None: + """Offline checkpoint inspection, also useful before resuming with an executor.""" + parser = argparse.ArgumentParser(description="Inspect a search board checkpoint (no model calls)") + parser.add_argument("--memory", required=True) + parser.add_argument("--search-id", required=True) + parser.add_argument("--head", required=True) + args = parser.parse_args() + board = Board.resume(open_store(args.memory), args.search_id, args.head, coordinator="inspect") + print(board.snapshot(), end="") + + +if __name__ == "__main__": + main() diff --git a/src/openultrasast/search/oracles.py b/src/openultrasast/search/oracles.py index ff2e9b9..ee2566a 100644 --- a/src/openultrasast/search/oracles.py +++ b/src/openultrasast/search/oracles.py @@ -11,12 +11,16 @@ import json import os import secrets import selectors +import shutil import sqlite3 import subprocess +import tempfile from collections.abc import Callable from dataclasses import dataclass +from html.parser import HTMLParser from pathlib import Path from typing import Protocol +from urllib.parse import quote @dataclass(frozen=True) @@ -136,18 +140,111 @@ class SSRFOracle(OutputOracle): self.process.stdout.close() -BrowserExecutor = Callable[[str, str], bool] +class BrowserExecutor: + """Run captured HTML with a verifier-owned alert canary. + + Payloads call alert("ousast-xss"). Reflection/escaped HTML cannot set the fresh + DOM marker. A restrictive CSP disables external resources and connections; + Chromium's sandbox stays enabled and each capture uses a fresh profile. + Launch failures are unavailable execution, never a negative observation. + """ + + def __init__(self, binary: str | None = None, *, timeout_seconds: float = 15) -> None: + selected = binary or shutil.which("chromium") or shutil.which("chromium-browser") + self.timeout_seconds = timeout_seconds + if not selected: + raise OSError("headless chromium unavailable") + self.binary: str = selected + + def __call__(self, document: str, nonce: str) -> bool: + marker = secrets.token_hex(24) + # Neither the marker nor its attribute appears in app output. The bootstrap + # source does appear in dump-dom, so inspect an actual DOM attribute only. + attribute = "data-proof-" + secrets.token_hex(12) + prelude = ( + '" + '" + ) + with tempfile.TemporaryDirectory(prefix="ousast-browser-") as profile: + # HTML can navigate even with CSP. A private network namespace, not + # browser flags, is the no-egress boundary. Expose runtime only. + sandbox = ["bwrap", "--unshare-all", "--die-with-parent", "--new-session"] + for path in ("/usr", "/bin", "/lib", "/lib64"): + if Path(path).exists(): + sandbox += ["--ro-bind", path, path] + sandbox += [ + "--dev", + "/dev", + "--proc", + "/proc", + "--tmpfs", + "/tmp", + "--bind", + profile, + "/profile", + "--setenv", + "HOME", + "/profile", + "--", + ] + try: + done = subprocess.run( + [ + *sandbox, + self.binary, + "--headless", + "--dump-dom", + "--disable-gpu", + "--disable-background-networking", + "--disable-extensions", + "--no-first-run", + "--no-default-browser-check", + "--disable-sync", + "--user-data-dir=/profile", + "--virtual-time-budget=1000", + "data:text/html;charset=utf-8," + quote(prelude + document), + ], + capture_output=True, + text=True, + timeout=self.timeout_seconds, + env={"PATH": "/usr/bin:/bin", "HOME": profile}, + ) + except subprocess.SubprocessError as exc: + raise OSError("headless chromium failed") from exc + if done.returncode or " None: + if tag == "html" and (attribute, marker) in attrs: + self.seen = True + + parser = Marker() + parser.feed(done.stdout) + return parser.seen + + +Browser = Callable[[str, str], bool] class XSSOracle(OutputOracle): - def __init__(self, executor: BrowserExecutor) -> None: + def __init__(self, executor: Browser) -> None: self.executor = executor def capture(self, output: str) -> None: self.seen |= self.executor(output, self.canary.nonce) -def oracle_for(family: str, browser: BrowserExecutor | None = None) -> Oracle | None: +def oracle_for(family: str, browser: Browser | None = None) -> Oracle | None: factories = {"path": PathOracle, "sql": SQLOracle, "command": CommandOracle, "ssrf": SSRFOracle} if family == "xss": return XSSOracle(browser) if browser else None diff --git a/tests/test_ax_batch.py b/tests/test_ax_batch.py index 3a877f2..58a112a 100644 --- a/tests/test_ax_batch.py +++ b/tests/test_ax_batch.py @@ -157,3 +157,21 @@ def test_bounded_lanes_oom_and_instrument_retry(tmp_path): assert dispatch([item("fail")], workload(), tmp_path, {"ax": failed}, lanes=("ax",)) assert attempts == [0, 1] + + +@pytest.mark.parametrize("step", ["reason", "explore", "verify"]) +def test_search_environment(step): + it = item() + it.extra_env = {"SEARCH_STEP": step, "SEARCH_ID": "search-1", "SEARCH_TASK_ID": "task-2"} + urls = {"SPEC_URL": "https://files.example/spec.dat", "RESULT_URL": "https://files.example/result.dat"} + doc = manifest("ousast-engine-search-1", workload(), it, urls) + assert {"name": "SEARCH_STEP", "value": step} in doc["spec"]["env"] + for key, value in [ + ("SEARCH_STEP", "shell"), + ("SEARCH_ID", "https://secret"), + ("SEARCH_TASK_ID", "x\nTOKEN=y"), + ("MODEL_API_KEY", "secret"), + ]: + it.extra_env = {key: value} + with pytest.raises(ValueError): + manifest("ousast-engine-search-1", workload(), it, urls) diff --git a/tests/test_search_board.py b/tests/test_search_board.py new file mode 100644 index 0000000..76c26f3 --- /dev/null +++ b/tests/test_search_board.py @@ -0,0 +1,39 @@ +import json + +import pytest + +from openultrasast.plane.memory import FileStore +from openultrasast.search.board import Board + + +def test_versions_provenance_claims_and_resume(tmp_path): + store = FileStore(tmp_path) + board = Board(store, "search-1", coordinator="host") + first = board.commit( + writer="host", facts=[dict(id="f1", text="sink", evidence_refs=["engine:1"], from_intents=[], step="triage", task_id="seed")] + ) + second = board.commit( + writer="host", + intents=[dict(id="i1", description="trace input", from_facts=["f1"], status="open", claimant=None, step="reason", task_id="r1")], + ) + assert first != second + assert json.loads(store.get_blob(board.prefix, first))["intents"] == [] + assert Board.resume(store, "search-1", second, coordinator="host").state == board.state + with pytest.raises(PermissionError): + board.commit(writer="worker", hints=[]) + state = board.state + state["facts"].clear() + assert board.state["facts"] + assert len(board.snapshot(max_bytes=256).encode()) <= 256 + with pytest.raises(ValueError): + board.commit(writer="host", end_reason="clean") + + +def test_bad_references_and_oversize_are_atomic(tmp_path): + board = Board(FileStore(tmp_path), "one", coordinator="host", max_bytes=1024) + before = board.head + with pytest.raises(ValueError): + board.commit(writer="host", facts=[dict(id="f", text="x", evidence_refs=[], from_intents=["missing"], step="explore", task_id="t")]) + with pytest.raises(ValueError): + board.commit(writer="host", hints=[dict(id="h", text="x" * 2048, step="human", task_id="operator")]) + assert board.head == before diff --git a/tests/test_search_browser.py b/tests/test_search_browser.py new file mode 100644 index 0000000..f2d2074 --- /dev/null +++ b/tests/test_search_browser.py @@ -0,0 +1,50 @@ +import html +import shutil +import subprocess + +import pytest + +from openultrasast.search.oracles import BrowserExecutor + + +def test_browser_launch_failure_is_not_negative(monkeypatch): + def fail(*args, **kwargs): + return subprocess.CompletedProcess(args, 1, "", "failed") + + monkeypatch.setattr(subprocess, "run", fail) + with pytest.raises(OSError): + BrowserExecutor(binary="/fake/chromium")('', "nonce") + + +@pytest.mark.skipif(not (shutil.which("chromium") or shutil.which("chromium-browser")), reason="chromium not installed") +def test_real_browser_executes_not_reflects(): + browser = BrowserExecutor() + payload = '' + try: + observed = browser(payload, "nonce") + except OSError as exc: + # A present binary is insufficient on restricted hosts. Only known + # host permission denials skip; ordinary browser failures remain failures. + if "Operation not permitted" in str(exc) or "Permission denied" in str(exc): + pytest.skip("host denies browser isolation/socket setup: " + str(exc)) + raise + assert observed + assert not browser(html.escape(payload), "nonce") + assert not browser("

nonce

", "nonce") + + +@pytest.mark.parametrize("executed", [False, True]) +def test_marker_is_dom_attribute_not_source_text(monkeypatch, executed): + import re + from urllib.parse import unquote + + def run(command, **kwargs): + assert "--headless" in command and "--dump-dom" in command + assert "--unshare-all" in command and "--no-sandbox" not in command + document = unquote(command[-1].split(",", 1)[1]) + attr, value = re.search(r'setAttribute\("([^"]+)","([^"]+)"\)', document).groups() + root = f'' if executed else "" + return subprocess.CompletedProcess(command, 0, root + document + "", "") + + monkeypatch.setattr(subprocess, "run", run) + assert BrowserExecutor(binary="/usr/bin/chromium")("

reflected nonce

", "nonce") is executed diff --git a/tests/test_search_budget.py b/tests/test_search_budget.py new file mode 100644 index 0000000..13b9823 --- /dev/null +++ b/tests/test_search_budget.py @@ -0,0 +1,58 @@ +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from openultrasast.plane.budget import BudgetExhausted, SpendBudget +from openultrasast.search.budget import SearchBudget + + +def test_reserve_settle_release(): + meter = SpendBudget(1) + reservation = meter.reserve(0.7) + with pytest.raises(BudgetExhausted): + meter.reserve(0.4) + reservation.settle(0.2) + assert meter.spent == 0.2 and meter.reserved == 0 + other = meter.reserve(0.8) + other.release() + assert meter.reserved == 0 + with pytest.raises(ValueError): + reservation.settle(0.1) + for bad in [-1, float("nan"), float("inf")]: + with pytest.raises(ValueError): + meter.reserve(bad) + + +def test_concurrent_reservations_cannot_overrun(): + meter = SpendBudget(1) + + def reserve(_): + try: + return meter.reserve(0.3) + except BudgetExhausted: + return None + + with ThreadPoolExecutor(max_workers=12) as pool: + reservations = list(pool.map(reserve, range(20))) + assert sum(r is not None for r in reservations) == 3 + assert meter.reserved <= 1 + + +def test_config_validation(): + assert SearchBudget().task_wall_seconds == 900 + for kwargs in [dict(retries=-1), dict(memory_bytes=0), dict(wall_seconds=float("nan"))]: + with pytest.raises(ValueError): + SearchBudget(**kwargs) + + +def test_settlement_bound_and_release_then_readmission(): + meter = SpendBudget(0.5) + reserved = meter.reserve(0.5) + with pytest.raises(ValueError, match="exceeds reservation"): + reserved.settle(0.6) + assert meter.reserved == 0.5 and meter.spent == 0 + reserved.release() + meter.reserve(0.5).settle(0.5) + assert meter.spent == 0.5 + with pytest.raises(BudgetExhausted): + meter.reserve(0.001) diff --git a/tests/test_search_coordinator.py b/tests/test_search_coordinator.py new file mode 100644 index 0000000..39d39d2 --- /dev/null +++ b/tests/test_search_coordinator.py @@ -0,0 +1,166 @@ +import pytest + +from openultrasast.plane.memory import FileStore +from openultrasast.search.board import Board +from openultrasast.search.budget import SearchBudget +from openultrasast.search.coordinator import Coordinator + + +def board(tmp_path): + return Board(FileStore(tmp_path), "search-1", coordinator="host") + + +def test_reason_explore_verify_and_resume(tmp_path): + calls = [] + + def execute(task): + calls.append(task) + assert task.limits["memory_bytes"] > 0 + if task.step == "reason": + return {"status": "ok", "cost_usd": 0.01, "intents": [{"description": "try input", "from_facts": []}]} + if task.step == "explore": + return {"status": "ok", "cost_usd": 0.02, "fact": {"text": "candidate", "evidence_refs": ["capture:1"], "demo": "demo:1"}} + return {"status": "ok", "outcome": "demonstrated", "evidence_refs": ["verify:1"]} + + b = board(tmp_path) + result = Coordinator(b, execute).run() + assert result["end_reason"] == "goal_met" + assert [t.step for t in calls] == ["reason", "explore", "verify"] + assert result["checkpoint"]["spent_usd"] == pytest.approx(0.03) + resumed = Board.resume(b.store, "search-1", b.head, coordinator="host") + assert Coordinator(resumed, execute).run() == result + assert len(calls) == 3 + + +def test_reason_claim_cannot_prove_goal(tmp_path): + result = Coordinator(board(tmp_path), lambda task: {"status": "ok", "cost_usd": 0, "complete": True, "facts": []}).run() + assert result["end_reason"] == "exhausted" + + +def test_refusal_before_dispatch_and_run_deadline(tmp_path): + calls = [] + result = Coordinator(board(tmp_path), calls.append, budget=SearchBudget(spend_usd=0.01)).run() + assert result["end_reason"] == "budget_spent" and calls == [] + times = iter([0, 100, 100, 100, 100]) + result = Coordinator(board(tmp_path / "two"), calls.append, budget=SearchBudget(wall_seconds=1), clock=lambda: next(times)).run() + assert result["end_reason"] == "budget_spent" and calls == [] + + +def test_retry_fallback_bounded_and_cost_released(tmp_path): + calls = [] + + def fail(task): + calls.append(task.lane) + return {"status": "sandbox_failure", "cost_usd": 0} + + state = Coordinator(board(tmp_path), fail, budget=SearchBudget(retries=1)).run() + assert calls == ["ax", "docker"] + assert state["end_reason"] == "exhausted" + assert state["checkpoint"]["retries"] == 1 + assert state["checkpoint"]["spent_usd"] == 0 + + +def test_claim_checkpoint_survives_interrupt_without_free_spend(tmp_path): + b = board(tmp_path) + + def interrupt(task): + raise KeyboardInterrupt + + with pytest.raises(KeyboardInterrupt): + Coordinator(b, interrupt).run() + checkpoint = b.head + resumed = Board.resume(b.store, "search-1", checkpoint, coordinator="host") + state = Coordinator(resumed, lambda task: {"status": "ok", "cost_usd": 0, "intents": []}).run() + assert state["checkpoint"]["spent_usd"] == 0.05 + assert state["checkpoint"]["tasks"] == 2 + + +def test_invalid_fact_fails_closed_and_records_failure(tmp_path): + def execute(task): + if task.step == "reason": + return {"status": "ok", "cost_usd": 0, "intents": [{"description": "x", "from_facts": []}]} + return {"status": "ok", "cost_usd": 0, "fact": {"text": "unbacked"}} + + state = Coordinator(board(tmp_path), execute, budget=SearchBudget(reason_rounds=1)).run() + assert not state["facts"] + assert state["intents"][0]["status"] == "concluded" + assert state["intents"][0]["failure"] == "invalid_result" + + +def test_cleanup_failure_halts_and_charges_unknown_spend(tmp_path): + b = board(tmp_path) + calls = [] + + def broken(task): + calls.append(task) + raise RuntimeError("AX cleanup failed; stop dispatching") + + with pytest.raises(RuntimeError, match="cleanup"): + Coordinator(b, broken).run() + assert len(calls) == 1 + assert b.state["checkpoint"]["spent_usd"] == 0.05 + assert b.state["checkpoint"]["pending"]["id"] == calls[0].id + + +def test_overclaimed_cost_stops_before_another_dispatch(tmp_path): + calls = [] + + def overrun(task): + calls.append(task) + return {"status": "ok", "cost_usd": 99, "intents": []} + + result = Coordinator(board(tmp_path), overrun).run() + assert result["end_reason"] == "budget_spent" + assert len(calls) == 1 + + +def test_claim_is_visible_in_worker_snapshot(tmp_path): + import yaml + + def execute(task): + if task.step == "reason": + return {"status": "ok", "cost_usd": 0, "intents": [{"description": "read", "from_facts": []}]} + intent = yaml.safe_load(task.snapshot)["intents"][0] + assert intent["status"] == "claimed" and intent["claimant"] == task.id + return {"status": "execution_failure", "cost_usd": 0} + + Coordinator(board(tmp_path), execute, budget=SearchBudget(reason_rounds=1)).run() + + +def test_batch_adapter_offline_checkpoint_and_step_dispatch(tmp_path, monkeypatch): + from dataclasses import asdict + from pathlib import Path + + monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[1])) + from openultrasast.search.coordinator import BatchExecutor, SearchTask + + calls = [] + + def lane(item, output, attempt): + import json + + spec = json.loads(item.inputs["SPEC_URL"]) + calls.append(spec) + assert item.extra_env["SEARCH_STEP"] == spec["step"] + return {"status": "ok", "cost_usd": 0, "intents": []} + + adapter = BatchExecutor( + image="registry/task@sha256:" + "a" * 64, command=("python3", "/worker.py"), output=tmp_path, lanes={"ax": lane} + ) + for index, step in enumerate(("reason", "explore", "verify")): + task = SearchTask(f"task-{index}", "search-1", step, "facts: []", {}, asdict(SearchBudget())) + assert adapter(task)["status"] == "ok" + assert adapter(task)["status"] == "ok" + assert len(calls) == 3 + + +def test_invalid_identifiers_refused_before_paths(tmp_path): + from dataclasses import asdict + + from openultrasast.search.coordinator import SearchTask + + for bad in ("../escape", "/tmp/escape", "bad\nNAME", ""): + with pytest.raises(ValueError): + Board(FileStore(tmp_path), bad, coordinator="host") + with pytest.raises(ValueError): + SearchTask(bad, "search-1", "reason", "", {}, asdict(SearchBudget())) diff --git a/tests/test_search_image.py b/tests/test_search_image.py new file mode 100644 index 0000000..f04a694 --- /dev/null +++ b/tests/test_search_image.py @@ -0,0 +1,18 @@ +from pathlib import Path + +import yaml + + +def test_image_matches_runner_and_publishes_digest(): + engine = Path("plane/Dockerfile.engine-task").read_text() + image = Path("plane/Dockerfile.search-task").read_text() + runner = "FROM ghcr.io/norandom/ax-task-runner:v0.3.1" + assert runner in engine and runner in image + for package in ("python3", "nodejs", "node-typescript", "php-cli", "sqlite3", "bubblewrap", "util-linux", "chromium"): + assert package in image + assert "17-jdk" in image and "--no-install-recommends" in image + assert "rm -rf /var/lib/apt/lists/" in image + workflow_text = Path(".github/workflows/engine-image.yml").read_text() + entries = yaml.safe_load(workflow_text)["jobs"]["image"]["strategy"]["matrix"]["include"] + assert {"package": "ousast-search-task", "dockerfile": "plane/Dockerfile.search-task"} in entries + assert "steps.build.outputs.digest" in workflow_text and "push: true" in workflow_text