diff --git a/benchmarks/ax/batch.py b/benchmarks/ax/batch.py index a2268e4..ccfd8db 100644 --- a/benchmarks/ax/batch.py +++ b/benchmarks/ax/batch.py @@ -715,7 +715,7 @@ class SearchExecutorTask: raise self.client = ObjectStoreExecutor(command_put, result_get) return self - except Exception as exc: + except BaseException as exc: self.__exit__(type(exc), exc, exc.__traceback__) raise @@ -739,6 +739,7 @@ class SearchExecutorTask: if self.submitted: try: self.lane.ax("delete", "task", self.name) + self.submitted = False except Exception as error: errors.append(exception_reason(error)) for key in self.keys: diff --git a/src/openultrasast/search/executor.py b/src/openultrasast/search/executor.py index 186992a..037535d 100644 --- a/src/openultrasast/search/executor.py +++ b/src/openultrasast/search/executor.py @@ -26,7 +26,7 @@ from collections.abc import Iterator, Mapping from contextlib import suppress from dataclasses import asdict, dataclass, field from pathlib import Path -from typing import Any, Protocol +from typing import Any, BinaryIO, Protocol from urllib.parse import urlsplit from ..sandbox import SandboxJob @@ -392,7 +392,7 @@ class InProcessExecutor: from .demo import preparation_command, validate_demo from .verify import _tree - from .verify_task import pack_checkout, validate_spec + from .verify_task import ArchiveLimit, bundle_ignore, pack_checkout, validate_spec if not self.task_boundary or not self.prepared_put_url: raise ValueError("preparation requires an executor task output") @@ -421,17 +421,29 @@ class InProcessExecutor: # regular archive members. No host/external files may be exported. _tree(self.repo, generated_links=True) _tree(products, generated_links=True) - shutil.copytree(self.repo, bundle / "checkout", symlinks=False) + shutil.copytree(self.repo, bundle / "checkout", symlinks=False, ignore=bundle_ignore) materialized = bundle / "materialized" - shutil.copytree(products, materialized, symlinks=False) + shutil.copytree(products, materialized, symlinks=False, ignore=bundle_ignore) shutil.rmtree(products) materialized.rename(products) (bundle / "spec.json").write_text(json.dumps(spec)) task_storage.check(task_storage.WORKSPACE if task_storage.WORKSPACE.is_dir() else bundle) - blob = pack_checkout(bundle) - self.phase = "result upload" - URLTransport().put(self.prepared_put_url, blob, self._remaining()) - return {"status": "ok", "archive_bytes": len(blob), "sha256": hashlib.sha256(blob).hexdigest()} + # The task entrypoint configures tempfile under disk-backed /workspace. + spool_root = task_storage.WORKSPACE if task_storage.WORKSPACE.is_dir() else bundle.parent + with tempfile.TemporaryFile(dir=spool_root) as archive: + self.phase = "build export" + try: + pack_checkout(bundle, archive) + except ArchiveLimit as exc: + return {"status": "could_not_build", "phase": self.phase, "reason": str(exc), "archive_bytes": exc.archive_bytes} + size = archive.tell() + task_storage.check(spool_root, additional_bytes=size) + archive.seek(0) + digest = hashlib.file_digest(archive, "sha256").hexdigest() + archive.seek(0) + self.phase = "result upload" + URLTransport().put(self.prepared_put_url, archive, self._remaining()) + return {"status": "ok", "archive_bytes": size, "sha256": digest} def _tree_bytes(self) -> int: total = 0 @@ -449,7 +461,7 @@ class InProcessExecutor: class Transport(Protocol): def get(self, url: str, timeout: float) -> bytes | None: ... - def put(self, url: str, data: bytes, timeout: float) -> None: ... + def put(self, url: str, data: bytes | BinaryIO, timeout: float) -> None: ... class URLTransport: @@ -472,15 +484,35 @@ class URLTransport: except urllib.error.HTTPError as exc: if exc.code == 404: return None - raise OSError(f"object read failed: HTTP {exc.code}") from None + raise - def put(self, url: str, data: bytes, timeout: float) -> None: - with self.opener.open( - urllib.request.Request(url, data=data, method="PUT", headers={"Content-Type": "application/json"}), timeout=timeout - ): + def put(self, url: str, data: bytes | BinaryIO, timeout: float) -> None: + headers = {"Content-Type": "application/json"} + if not isinstance(data, bytes): + headers["Content-Length"] = str(os.fstat(data.fileno()).st_size) + with self.opener.open(urllib.request.Request(url, data=data, method="PUT", headers=headers), timeout=timeout): pass +RETRY_SECONDS = 90.0 + + +def retry_mailbox(operation: Any, *, deadline: float | None = None) -> Any: + """Retry transient store failures; each attempt has a bounded socket timeout.""" + end = min(time.monotonic() + RETRY_SECONDS, deadline if deadline is not None else float("inf")) + delay = 0.2 + while True: + try: + return operation(max(0.001, min(10, end - time.monotonic()))) + except (OSError, TimeoutError) as exc: + if isinstance(exc, urllib.error.HTTPError) and not 500 <= exc.code < 600: + raise + if time.monotonic() >= end: + raise + time.sleep(min(delay, max(0, end - time.monotonic()))) + delay = min(delay * 2, 5) + + class ObjectStoreExecutor: def __init__(self, command_put_url: str, result_get_url: str, *, transport: Transport | None = None, poll_seconds: float = 0.2) -> None: for url in (command_put_url, result_get_url): @@ -492,26 +524,35 @@ class ObjectStoreExecutor: self.poll_seconds = poll_seconds self.seq = 0 self.uncertain = False + self.pending: Command | None = None self.isolation_mode = "remote-executor" def submit( self, name: str, args: dict[str, Any], *, timeout_seconds: float = 30, limits: dict[str, Any] | None = None ) -> dict[str, Any]: - if self.uncertain: - raise RuntimeError("previous command outcome uncertain; discard executor task") command = Command(self.seq + 1, name, args, timeout_seconds, limits or {}) + if self.pending is not None: + if (name, args, limits or {}) != (self.pending.name, self.pending.args, self.pending.limits): + raise RuntimeError("previous command outcome uncertain; retry the same command") + command = self.pending payload = command.validate() end = time.monotonic() + timeout_seconds self.uncertain = True + self.pending = copy.deepcopy(command) + retry_end = end + RETRY_SECONDS last_error = "" phase = "result upload" while time.monotonic() < end: try: # A lost PUT acknowledgement is retried with the exact same sequence. phase = "command upload" - self.transport.put(self.command_url, payload, max(0.001, end - time.monotonic())) - phase = "result read" - raw = self.transport.get(self.result_url, max(0.001, end - time.monotonic())) + before = time.monotonic() + try: + retry_mailbox(lambda timeout: self.transport.put(self.command_url, payload, timeout), deadline=retry_end) + phase = "result read" + raw = retry_mailbox(lambda timeout: self.transport.get(self.result_url, timeout), deadline=retry_end) + finally: + end = min(retry_end, end + time.monotonic() - before) if raw is not None: if len(raw) > MAX_RESULT_BYTES: raise ValueError("result too large") @@ -522,10 +563,14 @@ class ObjectStoreExecutor: raise ValueError("future result sequence") if result["seq"] == command.seq: self.seq, self.uncertain = command.seq, False + self.pending = None self.isolation_mode = result.get("isolation_mode", self.isolation_mode) return result except (OSError, TimeoutError) as exc: + if isinstance(exc, urllib.error.HTTPError) and not 500 <= exc.code < 600: + raise last_error = task_storage.exception_reason(exc) + break time.sleep(min(self.poll_seconds, max(0, end - time.monotonic()))) raise TimeoutError(f"executor {phase} deadline; {last_error or 'no result received'}") @@ -543,7 +588,7 @@ def serve( end = time.monotonic() + deadline_seconds while time.monotonic() < end: try: - raw = transport.get(command_get_url, min(10, max(0.001, end - time.monotonic()))) + raw = retry_mailbox(lambda timeout: transport.get(command_get_url, timeout), deadline=end) if raw is not None: if len(raw) > MAX_COMMAND_BYTES: raise ValueError("command too large") @@ -551,13 +596,16 @@ def serve( if command.seq != executor.seq and command.timeout_seconds > end - time.monotonic(): raise TimeoutError("command exceeds task lifetime") result = executor.execute(command) - transport.put(result_put_url, json.dumps(result).encode(), min(10, max(0.001, end - time.monotonic()))) + retry_mailbox( + lambda timeout, result=result: transport.put(result_put_url, json.dumps(result).encode(), timeout), deadline=end + ) if result.get("stopped"): return except (OSError, TimeoutError) as exc: if executor.last_result: executor.last_result["result_upload_reason"] = task_storage.exception_reason(exc) executor.last_result["result_upload_phase"] = "result upload" + raise time.sleep(min(poll_seconds, max(0, end - time.monotonic()))) raise TimeoutError("executor task deadline") diff --git a/src/openultrasast/search/verify.py b/src/openultrasast/search/verify.py index d73e24c..6f56c15 100644 --- a/src/openultrasast/search/verify.py +++ b/src/openultrasast/search/verify.py @@ -113,7 +113,9 @@ def _tree(root: Path, *, generated_links: bool = False) -> str: digest.update(str(path.relative_to(root)).encode()) digest.update(str(mode).encode()) if path.is_file(): - digest.update(path.read_bytes()) + with path.open("rb") as stream: + while block := stream.read(65536): + digest.update(block) if generated_links: visiting: set[Path] = set() visited: set[Path] = set() diff --git a/src/openultrasast/search/verify_task.py b/src/openultrasast/search/verify_task.py index ad3b223..77e3544 100644 --- a/src/openultrasast/search/verify_task.py +++ b/src/openultrasast/search/verify_task.py @@ -16,7 +16,7 @@ import urllib.error import urllib.request from dataclasses import asdict from pathlib import Path, PurePosixPath -from typing import Any, cast +from typing import Any, BinaryIO, cast from urllib.parse import urlsplit from . import task_storage @@ -47,78 +47,109 @@ def public_endpoint(url: str) -> str: return host -MAX_ARCHIVE_BYTES = 64 * 1024 * 1024 -MAX_CHECKOUT_BYTES = 256 * 1024 * 1024 +MAX_ARCHIVE_BYTES = 1024**3 +MAX_CHECKOUT_BYTES = 2 * 1024**3 MAX_FILES = 20000 MAX_SPEC_BYTES = 70000 MAX_RESULT_BYTES = 65536 -def pack_checkout(root: Path) -> bytes: +CACHE_NAMES = {".npm", ".pip", ".m2", ".composer", ".gradle", "__pycache__", ".git"} + + +def bundle_ignore(directory: str, names: list[str]) -> set[str]: + return {name for name in names if name in CACHE_NAMES or (Path(directory).name == "node_modules" and name == ".cache")} + + +class ArchiveLimit(ValueError): + def __init__(self, size: int, directories: dict[str, int]) -> None: + self.archive_bytes = size + largest = sorted(directories.items(), key=lambda item: (-item[1], item[0]))[:5] + super().__init__( + f"archive exceeds size limit: {size} bytes > {MAX_ARCHIVE_BYTES} bytes; top 5 directories: " + + ", ".join(f"{name} ({count} bytes)" for name, count in largest) + ) + + +def pack_checkout(root: Path, output: BinaryIO | None = None) -> bytes: root = Path(root) if root.is_symlink() or not root.is_dir(): raise ValueError("checkout must be a real directory") - buffer, total = io.BytesIO(), 0 + buffer = output if output is not None else io.BytesIO() + total, count = 0, 0 + directories: dict[str, int] = {} with tarfile.open(fileobj=buffer, mode="w:gz") as archive: - paths = (p for p in root.rglob("*") if ".git" not in p.relative_to(root).parts) - for index, path in enumerate(sorted(paths)): - mode = path.lstat().st_mode - if index >= MAX_FILES or not (stat.S_ISREG(mode) or stat.S_ISDIR(mode)): - raise ValueError("checkout contains too many files or special files") - total += path.stat().st_size if path.is_file() else 0 - if total > MAX_CHECKOUT_BYTES: - raise ValueError("checkout exceeds size limit") - archive.add(path, arcname=str(path.relative_to(root)), recursive=False) - data = buffer.getvalue() - if len(data) > MAX_ARCHIVE_BYTES: - raise ValueError("archive exceeds size limit") - return data + for parent, dirs, files in os.walk(root, followlinks=False): + ignored = bundle_ignore(parent, dirs + files) + dirs[:] = sorted(set(dirs) - ignored) + for name in sorted(dirs + [name for name in files if name not in ignored]): + path = Path(parent) / name + mode = path.lstat().st_mode + count += 1 + if count > MAX_FILES or not (stat.S_ISREG(mode) or stat.S_ISDIR(mode)): + raise ValueError("checkout contains too many files or special files") + size = path.stat().st_size if path.is_file() else 0 + total += size + if total > MAX_CHECKOUT_BYTES: + raise ValueError("checkout exceeds size limit") + for directory in path.relative_to(root).parents: + if str(directory) != ".": + directories[str(directory)] = directories.get(str(directory), 0) + size + archive.add(path, arcname=str(path.relative_to(root)), recursive=False) + if buffer.tell() > MAX_ARCHIVE_BYTES: + raise ArchiveLimit(buffer.tell(), directories) + return cast(io.BytesIO, buffer).getvalue() if output is None else b"" def extract_checkout(data: bytes, destination: Path) -> None: if len(data) > MAX_ARCHIVE_BYTES: raise ValueError("archive exceeds size limit") - # Bound expansion before tarfile parses PAX headers or allocates member data. - with gzip.GzipFile(fileobj=io.BytesIO(data)) as compressed: - expanded = compressed.read(MAX_CHECKOUT_BYTES + MAX_FILES * 4096 + 1) - if len(expanded) > MAX_CHECKOUT_BYTES + MAX_FILES * 4096: - raise ValueError("expanded checkout exceeds size limit") - destination.mkdir(parents=True) - total, seen = 0, set() - with tarfile.open(fileobj=io.BytesIO(expanded), mode="r:") as archive: - for index, member in enumerate(archive): - path = PurePosixPath(member.name) - if ( - index >= MAX_FILES - or path.is_absolute() - or ".." in path.parts - or not path.parts - or "\\" in member.name - or member.name in seen - or not (member.isfile() or member.isdir()) - ): - raise ValueError("unsafe checkout archive member") - seen.add(member.name) - total += member.size - if member.size < 0 or total > MAX_CHECKOUT_BYTES: - raise ValueError("expanded checkout exceeds size limit") - target = destination.joinpath(*path.parts) - if member.isdir(): - target.mkdir(parents=True, exist_ok=True) - else: - target.parent.mkdir(parents=True, exist_ok=True) - source = archive.extractfile(member) - if source is None: - raise ValueError("missing regular checkout member") - with source, target.open("xb") as output: - remaining = member.size - while remaining: - block = source.read(min(65536, remaining)) - if not block: - raise ValueError("truncated checkout member") - output.write(block) - remaining -= len(block) - target.chmod(0o755 if member.mode & 0o111 else 0o644) + # Bound expansion before tarfile parses PAX headers, using disk rather than + # allocating the entire expanded tree in worker RAM. + with tempfile.TemporaryFile() as expanded: + with gzip.GzipFile(fileobj=io.BytesIO(data)) as compressed: + maximum = MAX_CHECKOUT_BYTES + MAX_FILES * 4096 + while block := compressed.read(min(65536, maximum - expanded.tell() + 1)): + expanded.write(block) + if expanded.tell() > maximum: + raise ValueError("expanded checkout exceeds size limit") + expanded.seek(0) + destination.mkdir(parents=True) + total, seen = 0, set() + with tarfile.open(fileobj=expanded, mode="r:") as archive: + for index, member in enumerate(archive): + path = PurePosixPath(member.name) + if ( + index >= MAX_FILES + or path.is_absolute() + or ".." in path.parts + or not path.parts + or "\\" in member.name + or member.name in seen + or not (member.isfile() or member.isdir()) + ): + raise ValueError("unsafe checkout archive member") + seen.add(member.name) + total += member.size + if member.size < 0 or total > MAX_CHECKOUT_BYTES: + raise ValueError("expanded checkout exceeds size limit") + target = destination.joinpath(*path.parts) + if member.isdir(): + target.mkdir(parents=True, exist_ok=True) + else: + target.parent.mkdir(parents=True, exist_ok=True) + source = archive.extractfile(member) + if source is None: + raise ValueError("missing regular checkout member") + with source, target.open("xb") as output: + remaining = member.size + while remaining: + block = source.read(min(65536, remaining)) + if not block: + raise ValueError("truncated checkout member") + output.write(block) + remaining -= len(block) + target.chmod(0o755 if member.mode & 0o111 else 0o644) if not seen: raise ValueError("empty checkout") print(f"checkout_bytes={total} checkout_entries={len(seen)}", flush=True) diff --git a/src/openultrasast/search/worker.py b/src/openultrasast/search/worker.py index 032d5ea..d4831f6 100644 --- a/src/openultrasast/search/worker.py +++ b/src/openultrasast/search/worker.py @@ -355,6 +355,7 @@ class Worker: self.evidence = set() self.observations = [] malformed = 0 + plain_text = 0 for index in range(self.max_steps): messages[0]["content"] = ( EXPLORE_PROMPT @@ -367,7 +368,11 @@ class Worker: messages.append(message) calls = message.get("tool_calls", []) if not isinstance(calls, list) or not calls: - raise ValueError("explore must use tools or finish") + plain_text += 1 + if plain_text >= 2: + 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 finishes = [call for call in calls if isinstance(call, dict) and call.get("function", {}).get("name") == "finish"] if finishes: if len(calls) > 1: @@ -410,7 +415,7 @@ class Worker: raise ValueError("incomplete demonstration") schema = load_demo(self.demo) family = (yaml.safe_load(task.snapshot) or {}).get("family") - if family not in (None, "unknown"): + if family is not None and family != "unknown": validate_oracle(family, schema["oracle"]) fact["oracle"] = schema["oracle"] return {"fact": fact} diff --git a/tests/test_ax_batch.py b/tests/test_ax_batch.py index a4f3f1a..78f083f 100644 --- a/tests/test_ax_batch.py +++ b/tests/test_ax_batch.py @@ -177,7 +177,8 @@ def test_search_environment(step): manifest("ousast-engine-search-1", workload(), it, urls) -def test_search_executor_task_links_and_cleanup(): +@pytest.mark.parametrize("failure", [None, TimeoutError, KeyboardInterrupt]) +def test_search_executor_task_links_and_cleanup(monkeypatch, failure): from types import SimpleNamespace from benchmarks.ax.batch import SearchExecutorTask @@ -209,13 +210,22 @@ def test_search_executor_task_links_and_cleanup(): clock=__import__("time").monotonic, pause=lambda _: None, ) - with SearchExecutorTask(lane, b"archive", deadline=60) as session: + from contextlib import nullcontext + + with pytest.raises(failure) if failure else nullcontext(), SearchExecutorTask(lane, b"archive", deadline=60) as session: assert session.name.startswith("ousast-engine-search-exec-") assert session.client.command_url.endswith("?put") doc = calls[0][1]["manifest"] assert doc["spec"]["command"][2] == "openultrasast.search.executor" assert {r["name"] for r in doc["spec"]["env"]} == {"COMMAND_GET_URL", "EXECUTOR_RESULT_PUT_URL", "REPO_URL", "PREPARED_PUT_URL"} assert "API_KEY" not in json.dumps(doc) + if failure: + + def submit(*args, **kwargs): + raise failure("transport interrupted") + + monkeypatch.setattr(session.client, "submit", submit) + session.client.submit("list_files", {}) assert calls[-1][0][:2] == ("delete", "task") assert len(lane.store.deleted) == 4 diff --git a/tests/test_search_executor.py b/tests/test_search_executor.py index e4b0b79..d4a542e 100644 --- a/tests/test_search_executor.py +++ b/tests/test_search_executor.py @@ -77,7 +77,7 @@ def test_mailbox_timeout(): with pytest.raises(TimeoutError): client.submit("list_files", {}, timeout_seconds=0.01) with pytest.raises(RuntimeError, match="uncertain"): - client.submit("list_files", {}) + client.submit("stop", {}) def test_serve_retries_result_without_reexecuting(tmp_path): @@ -96,7 +96,7 @@ def test_serve_retries_result_without_reexecuting(tmp_path): assert json.loads(data)["stopped"] transport = Transport() - serve(executor, "command", "result", transport=transport, deadline_seconds=0.2, poll_seconds=0.001) + serve(executor, "command", "result", transport=transport, deadline_seconds=1, poll_seconds=0.001) assert transport.writes == 2 @@ -316,3 +316,123 @@ def test_guard_io_failure_after_child_exit_is_not_success(tmp_path, monkeypatch) assert result["phase"] == "scratch guard" assert result["exit_code"] is not None assert "cache cannot be read" in result["reason"] + + +@pytest.mark.parametrize("side", ["host", "task"]) +@pytest.mark.parametrize("method", ["put", "get"]) +@pytest.mark.parametrize("error", ["timeout", "connection", "503"]) +def test_transient_mailbox_failures_retry_same_sequence(tmp_path, monkeypatch, side, method, error): + import urllib.error + + from openultrasast.search import executor as module + + now = [0.0] + monkeypatch.setattr(module.time, "monotonic", lambda: now[0]) + monkeypatch.setattr(module.time, "sleep", lambda delay: now.__setitem__(0, now[0] + delay)) + app = InProcessExecutor(tmp_path) + failures, payloads, results = [], [], [] + + def fail(): + failures.append(now[0]) + if len(failures) <= 4: + if error == "503": + raise urllib.error.HTTPError("https://store.example", 503, "unavailable", {}, None) + raise (TimeoutError if error == "timeout" else ConnectionError)("temporary store failure") + + class Flaky: + def put(self, url, data, timeout): + payloads.append(data) + if side == "host": + results[:] = [app.execute(Command(**json.loads(data)))] + if method == "put": + fail() + + def get(self, url, timeout): + if method == "get": + fail() + if side == "host": + return json.dumps(results[-1]).encode() + return Command(1, "stop", {}, 1).validate() + + if side == "host": + client = ObjectStoreExecutor("https://store.example/command", "https://store.example/result", transport=Flaky()) + assert client.submit("stop", {}, timeout_seconds=1)["stopped"] + assert client.seq == 1 and not client.uncertain + else: + serve(app, "command", "result", transport=Flaky(), deadline_seconds=100) + assert app.seq == 1 and app.stopped + assert len(set(payloads)) == 1 + assert len(failures) == 5 and now[0] > 1 + + +def test_retry_budget_is_90_seconds_and_permanent_http_errors_fail_fast(monkeypatch): + import urllib.error + + from openultrasast.search import executor as module + + now, calls = [0.0], [] + monkeypatch.setattr(module.time, "monotonic", lambda: now[0]) + monkeypatch.setattr(module.time, "sleep", lambda delay: now.__setitem__(0, now[0] + delay)) + + def fail(timeout): + calls.append(timeout) + raise TimeoutError("still unavailable") + + with pytest.raises(TimeoutError): + module.retry_mailbox(fail) + assert now[0] == 90 and all(0 < value <= 10 for value in calls) + + def forbidden(timeout): + raise urllib.error.HTTPError("https://store.example", 403, "forbidden", {}, None) + + with pytest.raises(urllib.error.HTTPError): + module.retry_mailbox(forbidden) + assert now[0] == 90 + + +def test_uncertain_command_can_be_resubmitted_without_changing_seq(tmp_path): + class Lost(Mailbox): + available = False + + def put(self, url, data, timeout): + self.writes += 1 + self.result = json.dumps(self.executor.execute(Command(**json.loads(data)))).encode() + + def get(self, url, timeout): + return self.result if self.available else None + + (tmp_path / "input").write_text("original") + mailbox = Lost(InProcessExecutor(tmp_path)) + client = ObjectStoreExecutor("https://store.example/command", "https://store.example/result", transport=mailbox, poll_seconds=0.001) + with pytest.raises(TimeoutError): + client.submit("read_file", {"path": "input"}, timeout_seconds=0.01) + (tmp_path / "input").write_text("changed") + mailbox.available = True + assert client.submit("read_file", {"path": "input"})["text"] == "original" + assert client.seq == 1 + + +def test_url_transport_uploads_file_without_buffering(tmp_path): + from contextlib import nullcontext + + from openultrasast.search.executor import URLTransport + + path = tmp_path / "archive" + path.write_bytes(b"synthetic archive" * 10000) + transport = URLTransport() + seen = {} + + class Opener: + def open(self, request, timeout): + assert request.get_method() == "PUT" + assert not isinstance(request.data, bytes) + seen["length"] = int(request.get_header("Content-length")) + seen["bytes"] = 0 + while block := request.data.read(8192): + seen["bytes"] += len(block) + return nullcontext() + + transport.opener = Opener() + with path.open("rb") as stream: + transport.put("https://files.example/archive", stream, 10) + assert seen["length"] == seen["bytes"] == path.stat().st_size diff --git a/tests/test_search_pilot_run.py b/tests/test_search_pilot_run.py index bc8d97d..e257847 100644 --- a/tests/test_search_pilot_run.py +++ b/tests/test_search_pilot_run.py @@ -131,7 +131,7 @@ def test_worker_failure_is_not_no_evidence(tmp_path): 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"] + assert "second plain-text reply:" in record["failures"][0]["reason"] def test_shared_budget_reserved_and_settled_on_error(tmp_path): @@ -500,3 +500,41 @@ def test_pilot_retains_command_diagnostics(tmp_path, phase): assert "line-29" in failure["stderr"] and "line-0\n" not in failure["stderr"] assert "hidden" not in failure["stderr"] and "https://" not in failure["stderr"] assert len(failure["stderr"]) <= 1500 + + +@pytest.mark.parametrize("failure_at", ["acquire", "body"]) +def test_pilot_executor_deletes_task_when_submit_times_out(tmp_path, monkeypatch, failure_at): + import time + from types import SimpleNamespace + + from openultrasast.search.executor import ObjectStoreExecutor + + pilot = make_pilot(tmp_path) + deleted, calls = [], [] + store = SimpleNamespace( + _put=lambda *a: None, + _delete=deleted.append, + presign_get=lambda key, expires: "https://files.example/" + key, + presign_put=lambda key, expires: "https://files.example/" + key, + ) + pilot.lane = SimpleNamespace( + store=store, + host="files.example", + args=SimpleNamespace(atespace="default"), + workload=SimpleNamespace(image="registry/task@sha256:" + "a" * 64), + ax=lambda *a, **kw: calls.append((a, kw)), + clock=time.monotonic, + pause=lambda _: None, + ) + + def submit(self, name, args, **kwargs): + if failure_at == "acquire" or name == "list_files": + raise TimeoutError("executor command upload deadline") + return {"exit_code": 0, "input_bytes": 10} + + monkeypatch.setattr(ObjectStoreExecutor, "submit", submit) + with pytest.raises(TimeoutError, match="upload deadline"), pilot.executor("a" * 40, deadline=60) as session: + session.client.submit("list_files", {}) + assert calls[-1][0][:2] == ("delete", "task") + assert len(deleted) == 4 + assert pilot.record["executor_task_metrics"][-1]["end"] == "execution_failure" diff --git a/tests/test_search_task_storage.py b/tests/test_search_task_storage.py index 8359fd9..da60d91 100644 --- a/tests/test_search_task_storage.py +++ b/tests/test_search_task_storage.py @@ -75,7 +75,9 @@ def test_executor_exports_built_input_and_verifier_needs_no_build(workspace, mon blobs = [] app = executor.InProcessExecutor(repo, task_boundary=True) app.prepared_put_url = "https://files.example/output" - monkeypatch.setattr(executor.URLTransport, "put", lambda self, url, data, timeout: blobs.append(data)) + monkeypatch.setattr( + executor.URLTransport, "put", lambda self, url, data, timeout: blobs.append(data.read() if hasattr(data, "read") else data) + ) result = app.submit("prepare_verification", spec()) assert result["status"] == "ok", result destination = workspace / "unpacked" @@ -140,7 +142,9 @@ def test_preparation_builds_before_export_and_exports_products(workspace, monkey app = executor.InProcessExecutor(repo, task_boundary=True) app.prepared_put_url = "https://files.example/output" blobs, commands = [], [] - monkeypatch.setattr(executor.URLTransport, "put", lambda self, url, data, timeout: blobs.append(data)) + monkeypatch.setattr( + executor.URLTransport, "put", lambda self, url, data, timeout: blobs.append(data.read() if hasattr(data, "read") else data) + ) def build(argv, timeout, limits, **kwargs): commands.append(argv) @@ -246,7 +250,9 @@ def test_preparation_uses_project_gradle_wrapper(workspace, monkeypatch, project app = executor.InProcessExecutor(repo, task_boundary=True) app.prepared_put_url = "https://files.example/output" blobs = [] - monkeypatch.setattr(executor.URLTransport, "put", lambda self, url, data, timeout: blobs.append(data)) + monkeypatch.setattr( + executor.URLTransport, "put", lambda self, url, data, timeout: blobs.append(data.read() if hasattr(data, "read") else data) + ) request = spec() request["demo"]["build"] = {"recipe": "gradle", "arguments": [project]} result = app.submit("prepare_verification", request) @@ -302,7 +308,9 @@ def test_prepare_build_virtual_memory(workspace, monkeypatch, memory_bytes): app = executor.InProcessExecutor(repo, task_boundary=True) app.prepared_put_url = "https://files.example/output" blobs = [] - monkeypatch.setattr(executor.URLTransport, "put", lambda self, url, data, timeout: blobs.append(data)) + monkeypatch.setattr( + executor.URLTransport, "put", lambda self, url, data, timeout: blobs.append(data.read() if hasattr(data, "read") else data) + ) request = spec() request["demo"]["build"] = {"recipe": "npm", "arguments": ["."]} limits = {} if memory_bytes is None else {"memory_bytes": memory_bytes} @@ -440,7 +448,9 @@ def test_prepare_preserves_large_config_and_task_verifier_applies_it(workspace, blobs = [] app = executor.InProcessExecutor(repo, task_boundary=True) app.prepared_put_url = "https://files.example/output" - monkeypatch.setattr(executor.URLTransport, "put", lambda self, url, data, timeout: blobs.append(data)) + monkeypatch.setattr( + executor.URLTransport, "put", lambda self, url, data, timeout: blobs.append(data.read() if hasattr(data, "read") else data) + ) result = app.submit("prepare_verification", request) assert result["status"] == "ok", result destination = workspace / "unpacked" @@ -454,3 +464,64 @@ def test_prepare_preserves_large_config_and_task_verifier_applies_it(workspace, observation = verify_side_task(Side(destination / "checkout", products=destination / "products"), demo, "path") assert observation.outcome == "observed", observation assert len(observation.runs) == 1 + + +def test_large_export_spools_to_workspace_and_excludes_caches(workspace, monkeypatch): + import hashlib + import tarfile + + repo = workspace / "checkout" + repo.mkdir() + (repo / "app.py").write_text("print('safe')") + # Incompressible input exceeds the former 64 MiB archive cap. + with (repo / "large.bin").open("wb") as stream: + for _ in range(129): + stream.write(os.urandom(1024**2)) + for directory in [".npm", ".pip", ".m2", ".composer", ".gradle", "__pycache__", ".git", "node_modules/.cache"]: + path = repo / "nested" / directory + path.mkdir(parents=True) + (path / "excluded").write_bytes(b"cache") + seen = {} + + def upload(self, url, data, timeout): + assert not isinstance(data, bytes) + assert str(workspace) in os.readlink(f"/proc/self/fd/{data.fileno()}") + seen["size"] = os.fstat(data.fileno()).st_size + seen["sha256"] = hashlib.file_digest(data, "sha256").hexdigest() + data.seek(0) + with tarfile.open(fileobj=data, mode="r:gz") as archive: + names = archive.getnames() + assert "checkout/large.bin" in names + assert not any(name.endswith("excluded") for name in names) + data.seek(0) + + monkeypatch.setattr(executor.URLTransport, "put", upload) + app = executor.InProcessExecutor(repo, task_boundary=True) + app.prepared_put_url = "https://files.example/output" + result = app.submit("prepare_verification", spec(), timeout_seconds=60) + assert result["status"] == "ok", result + assert 128 * 1024**2 < result["archive_bytes"] == seen["size"] + assert result["sha256"] == seen["sha256"] + assert verify_task.MAX_ARCHIVE_BYTES == 1024**3 + + +def test_export_over_cap_reports_size_and_five_relative_directories(workspace, monkeypatch): + repo = workspace / "checkout" + repo.mkdir() + (repo / "app.py").write_text("print('safe')") + for i in range(6): + directory = repo / f"dir{i}" + directory.mkdir() + (directory / "data").write_bytes(os.urandom(1000 * (i + 1))) + monkeypatch.setattr(verify_task, "MAX_ARCHIVE_BYTES", 1000) + monkeypatch.setattr(executor.URLTransport, "put", lambda *a: pytest.fail("oversized upload")) + app = executor.InProcessExecutor(repo, task_boundary=True) + app.prepared_put_url = "https://files.example/output" + result = app.submit("prepare_verification", spec()) + assert result["status"] == "could_not_build" + assert str(result["archive_bytes"]) + " bytes > 1000 bytes" in result["reason"] + assert str(workspace) not in result["reason"] + assert "top 5 directories: checkout (" in result["reason"] + for i in range(2, 6): + assert f"checkout/dir{i} (" in result["reason"] + assert "dir0" not in result["reason"] and "dir1" not in result["reason"] diff --git a/tests/test_search_verify_transport.py b/tests/test_search_verify_transport.py index b0f1186..f4c1e20 100644 --- a/tests/test_search_verify_transport.py +++ b/tests/test_search_verify_transport.py @@ -77,7 +77,12 @@ def test_entry_reads_input_and_scrubs_environment(tmp_path, monkeypatch, capsys) (bundle / "products").mkdir() (bundle / "spec.json").write_text(json.dumps({"demo": demo(), "family": "injection", "timeout_seconds": 2})) blob = entry.pack_checkout(bundle) - monkeypatch.setattr(entry, "download", lambda url, limit: blob) + + def download(url, limit): + assert limit == 1024**3 + return blob + + monkeypatch.setattr(entry, "download", download) outputs = [] monkeypatch.setattr(entry, "upload", lambda url, data: outputs.append(json.loads(data))) diff --git a/tests/test_search_worker.py b/tests/test_search_worker.py index 5fc294d..e17d55b 100644 --- a/tests/test_search_worker.py +++ b/tests/test_search_worker.py @@ -457,3 +457,27 @@ def test_finish_wins_over_other_calls_without_executing_them(tmp_path, finish_fi assert result["fact"]["text"] == "Confirmed app source" assert worker.metrics["warnings"] == 1 assert worker.evidence == {"tool:1"} + + +def test_plain_text_gets_one_reminder_then_can_finish(tmp_path): + model = StubModel([{"content": "I will inspect"}, tool("finish", failure="no evidence")]) + result = Worker(model, repo=tmp_path, demo=tmp_path / "demo", max_steps=2)(task("explore")) + assert result["failure"] == "no evidence" + assert len(model.calls) == 2 + assert model.calls[1]["messages"][-1] == {"role": "user", "content": "Use a tool or call finish; plain text is not accepted."} + + +def test_second_plain_text_records_first_300_characters(tmp_path): + text = "second response " * 40 + model = StubModel([{"content": "first"}, {"content": text}]) + result = Worker(model, repo=tmp_path, demo=tmp_path / "demo", max_steps=3)(task("explore")) + assert result["status"] == "execution_failure" + assert result["failure"] == "second plain-text reply: " + text[:300] + assert len(model.calls) == 2 + + +def test_plain_text_consumes_step(tmp_path): + model = StubModel([{"content": "first"}]) + 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