merge(search-with-proof): archive, text replies, transport retries [skip ci]

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

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:

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)
+            # 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, blob, self._remaining())
-            return {"status": "ok", "archive_bytes": len(blob), "sha256": hashlib.sha256(blob).hexdigest()}
+                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()))
+                before = time.monotonic()
+                try:
+                    retry_mailbox(lambda timeout: self.transport.put(self.command_url, payload, timeout), deadline=retry_end)
                     phase = "result read"
-                raw = self.transport.get(self.result_url, max(0.001, end - time.monotonic()))
+                    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")

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()

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,45 +47,76 @@ 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)):
+        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
-            if index >= MAX_FILES or not (stat.S_ISREG(mode) or stat.S_ISDIR(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")
-            total += path.stat().st_size if path.is_file() else 0
+                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)
-    data = buffer.getvalue()
-    if len(data) > MAX_ARCHIVE_BYTES:
-        raise ValueError("archive exceeds size limit")
-    return data
+    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.
+    # 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:
-        expanded = compressed.read(MAX_CHECKOUT_BYTES + MAX_FILES * 4096 + 1)
-    if len(expanded) > MAX_CHECKOUT_BYTES + MAX_FILES * 4096:
+            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=io.BytesIO(expanded), mode="r:") as archive:
+        with tarfile.open(fileobj=expanded, mode="r:") as archive:
             for index, member in enumerate(archive):
                 path = PurePosixPath(member.name)
                 if (

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}

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
 

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

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"

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

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)))
 

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