From db46250d852816687ae943c306cf93b319d13706 Mon Sep 17 00:00:00 2001 From: Lucien <164857476+dakjdakd@users.noreply.github.com> Date: Mon, 12 Oct 2026 22:16:33 +0800 Subject: [PATCH] fix(opensandbox): mirror thread outputs before warm reuse (#6699) * fix(opensandbox): mirror thread outputs before warm reuse * docs(opensandbox): scope output synchronization guidance locally * docs(opensandbox): clarify portable output name requirements * fix(opensandbox): isolate invalid filenames and type lifecycle closure * fix(opensandbox): mirror complete records after find walk errors * fix(opensandbox): reject ambiguous output path aliases --------- Co-authored-by: Willem Jiang --- README.md | 9 + .../deerflow/community/opensandbox/AGENTS.md | 27 + .../deerflow/community/opensandbox/README.md | 36 ++ .../community/opensandbox/provider.py | 199 +++++- .../deerflow/community/opensandbox/sandbox.py | 25 +- .../deerflow/sandbox/acquire_serialization.py | 10 +- backend/tests/test_acquire_serialization.py | 14 +- backend/tests/test_agent_guidance_check.py | 1 + backend/tests/test_opensandbox_provider.py | 578 ++++++++++++++++++ 9 files changed, 880 insertions(+), 19 deletions(-) create mode 100644 backend/packages/harness/deerflow/community/opensandbox/AGENTS.md diff --git a/README.md b/README.md index f1ab1d2165..a22f82d55e 100644 --- a/README.md +++ b/README.md @@ -792,6 +792,15 @@ report `Exit Code: 124`; unsuccessful health checks cannot reclaim a warm sandbo Health probes tolerate login-shell output around the `ok` line, and failures log the sandbox ID and probe output before replacing the sandbox. +The optional OpenSandbox provider mirrors generated output files to local thread +storage at turn completion for Gateway previews and downloads; see its +[synchronization limits and failure behavior](backend/packages/harness/deerflow/community/opensandbox/README.md#lifecycle-and-contract). +An invalid UTF-8 filename is skipped without blocking other outputs, and +same-size content edits are still copied on release. +Outputs with case- or Unicode-equivalent file or directory names are skipped +before copying, and aliases of existing local paths are rejected, preserving +previous complete copies. + BoxLite shutdown rejects late VM registration and keeps its SDK loop open while in-flight acquisitions finish. If they cannot drain within five seconds, shutdown fails with resources still owned and can be retried. diff --git a/backend/packages/harness/deerflow/community/opensandbox/AGENTS.md b/backend/packages/harness/deerflow/community/opensandbox/AGENTS.md new file mode 100644 index 0000000000..8355032d2d --- /dev/null +++ b/backend/packages/harness/deerflow/community/opensandbox/AGENTS.md @@ -0,0 +1,27 @@ +### OpenSandbox output synchronization + +Release mirrors scoped regular output files to `Paths.sandbox_outputs_dir()` +before parking the client. Hold the same scope serializer as acquire and retain +the active entry during transfer so reuse and warm reaping cannot overlap it. +Never repool a client invalidated or untracked during synchronization. + +The bounded listing uses base64-encoded NUL frames to survive line-oriented SSE +events. POSIX pipelines must report every stage's exit status explicitly. +`find` exit 1 permits complete records with a warning only after size/framing +validation; failed head/encoder stages and truncated listings still fail closed. +Native downloads enforce the remaining byte budget and cooperative deadline, +closing the stream on failure. Publish via an exclusively created temporary file and +atomic replacement; reject traversal and existing linked host destinations. +Preflight the entire accepted listing for case-folded, NFC-equivalent component +aliases before any download, including records beyond the transfer cap. Skip all +paths under ambiguous components. Reject differently spelled local aliases both +before transfer and before publication; preserve previous complete copies. +Keep failed synchronization best effort, preserve prior complete copies, and +do not remove remotely deleted files. Cover the real listing shell and release +to Gateway artifact reader with offline transports in +`tests/test_opensandbox_provider.py`. +Decode each NUL-framed filename separately; invalid UTF-8 records consume the +candidate budget but must not prevent valid outputs from being mirrored. +Portable reserved-name checks include COM/LPT superscript digits and console +device names without `PureWindowsPath.is_reserved`. Catch only the serializer's +typed closure exception after verifying the client has been untracked. diff --git a/backend/packages/harness/deerflow/community/opensandbox/README.md b/backend/packages/harness/deerflow/community/opensandbox/README.md index c0579b314b..05877a68b6 100644 --- a/backend/packages/harness/deerflow/community/opensandbox/README.md +++ b/backend/packages/harness/deerflow/community/opensandbox/README.md @@ -44,6 +44,42 @@ that itself starts with `$` reaches the sandbox intact. The provider derives a stable local ID from `(user_id, thread_id)`. A released sandbox enters an in-process warm pool and only the same scope may reclaim it. +Before parking a user/thread-scoped sandbox, the provider copies regular files +from `/mnt/user-data/outputs` to that owner's local thread outputs directory. +This makes generated artifacts available to Gateway previews and downloads after +the turn finishes. Nested paths and binary content are preserved; uploads and +workspace files are not mirrored. The same scope waits for this copy before its +next acquire. + +Output synchronization is best effort: failures log a warning and preserve any +previous complete local copy. Each pass processes at most 2,000 files, a 512 KiB +listing and 512 MiB of downloaded content, with a 100 MiB per-file limit. An +oversized listing skips the pass; a failed or oversized download stops the pass +because a partial stream cannot be charged accurately to the remaining budget. +A 120-second +cooperative deadline is checked between files and download chunks; an active SDK +read is bounded by `request_timeout`, so this is not a hard wall-clock limit. +The listing requires `find`, `head`, `base64` and `mktemp` in the remote image. +If `find` reports a traversal error, complete NUL-framed records are still +mirrored with a warning; oversized or incomplete listings and failed pipeline +stages skip the pass. +Remote links are not followed and linked local destinations are rejected. +Mirrored names must also be portable host paths: traversal, backslashes, colons, +Windows reserved components and components ending in a space or dot are skipped +with a warning, including when the Gateway runs on Unix. +The entire accepted listing is checked before any download for names that differ +only by case or Unicode normalization, including directory components and records +beyond the file limit. All outputs under an ambiguous component are skipped, on +every host, so one remote report cannot overwrite another on a case-insensitive +volume. A path that aliases a differently spelled existing local file or directory +is also skipped; previous complete copies are preserved. Unambiguous outputs +continue to be mirrored. +Files deleted remotely are not removed from the local outputs directory. +Non-UTF-8 filename records are skipped individually with a warning and still +count against the file limit. Each release revisits retained outputs, including +unchanged files; equal file size does not prove equal content. Large accumulated +output directories may reach the pass limits before all files are mirrored. + Create returns only after the SDK readiness check and DeerFlow's `/mnt/user-data/{workspace,uploads,outputs}` bootstrap succeed. A bootstrap failure explicitly destroys the newly created remote sandbox. diff --git a/backend/packages/harness/deerflow/community/opensandbox/provider.py b/backend/packages/harness/deerflow/community/opensandbox/provider.py index 5923130019..075b166710 100644 --- a/backend/packages/harness/deerflow/community/opensandbox/provider.py +++ b/backend/packages/harness/deerflow/community/opensandbox/provider.py @@ -3,26 +3,32 @@ from __future__ import annotations import atexit +import base64 import ipaddress import logging import math import os +import shlex +import stat +import tempfile import threading import time +import unicodedata import uuid from datetime import timedelta from functools import partial +from pathlib import Path, PurePosixPath from typing import TYPE_CHECKING, Any from urllib.parse import urlsplit from deerflow.config import get_app_config -from deerflow.sandbox.acquire_serialization import AcquireSerializer +from deerflow.sandbox.acquire_serialization import AcquireSerializer, AcquireSerializerClosed from deerflow.sandbox.identity import derive_sandbox_scope_token from deerflow.sandbox.sandbox import Sandbox, _validate_extra_env from deerflow.sandbox.sandbox_provider import SandboxProvider from ..warm_pool_lifecycle import WarmPoolLifecycleMixin -from .sandbox import OpenSandboxSandbox, format_execution +from .sandbox import _MAX_DOWNLOAD_SIZE, OpenSandboxSandbox, execution_stdout, format_execution if TYPE_CHECKING: from opensandbox.config.connection_sync import ConnectionConfigSync @@ -38,6 +44,7 @@ DEFAULT_SANDBOX_TIMEOUT = 4 * 60 * 60 DEFAULT_COMMAND_TIMEOUT = 10 * 60 _BOOTSTRAP_TIMEOUT = 30.0 _BOOTSTRAP_COMMAND = "mkdir -p /mnt/user-data/workspace /mnt/user-data/uploads /mnt/user-data/outputs" +_WINDOWS_RESERVED_NAMES = frozenset({"CON", "PRN", "AUX", "NUL", "CONIN$", "CONOUT$"} | {f"{prefix}{digit}" for prefix in ("COM", "LPT") for digit in "123456789¹²³"}) def _uses_insecure_remote_http(domain: Any, protocol: str) -> bool: @@ -75,6 +82,10 @@ class OpenSandboxProvider(WarmPoolLifecycleMixin[OpenSandboxSandbox], SandboxPro uses_thread_data_mounts = False needs_upload_permission_adjustment = True _idle_checker_thread_name = "opensandbox-idle-reaper" + _MAX_SYNC_FILES = 2000 + _MAX_SYNC_TOTAL_BYTES = 512 * 1024 * 1024 + _MAX_SYNC_LISTING_BYTES = 512 * 1024 + _SYNC_DEADLINE_SECONDS = 120.0 def __init__(self) -> None: self._lock = threading.Lock() @@ -354,20 +365,190 @@ class OpenSandboxProvider(WarmPoolLifecycleMixin[OpenSandboxSandbox], SandboxPro return self._sandboxes.get(sandbox_id) def release(self, sandbox_id: str) -> None: + # Same scope lock as acquire: no next turn may reuse a client while its + # previous outputs are still being mirrored. Keep it active until sync + # finishes so the warm-pool reaper cannot destroy it during transfer. with self._lock: - sandbox = self._sandboxes.pop(sandbox_id, None) + if sandbox_id not in self._sandboxes: + return + try: + with self._acquire_serializer.hold(sandbox_id): + self._release_scope_locked(sandbox_id) + except AcquireSerializerClosed: + # reset/shutdown can close the serializer between the existence + # check and checkout. Preserve release's no-op after termination. + with self._lock: + untracked = sandbox_id not in self._sandboxes + if not untracked: + raise + + def _release_scope_locked(self, sandbox_id: str) -> None: + with self._lock: + sandbox = self._sandboxes.get(sandbox_id) + scope = next((key for key, value in self._thread_sandboxes.items() if value == sandbox_id), None) + if sandbox is None: + return + if scope is not None and scope[0]: + try: + self._sync_outputs_to_host(sandbox, user_id=scope[0], thread_id=scope[1]) + except Exception: + logger.warning("Failed to synchronize OpenSandbox outputs for %s", sandbox_id, exc_info=True) + with self._lock: + # Terminal transfer failures or shutdown may already untrack + # this client. Never resurrect it in the warm pool. + if self._sandboxes.get(sandbox_id) is not sandbox: + return + self._sandboxes.pop(sandbox_id) for key in [key for key, value in self._thread_sandboxes.items() if value == sandbox_id]: self._thread_sandboxes.pop(key, None) - if sandbox is None: - return - if self._shutdown_called: - destroy_after_unlock = True - else: + destroy_after_unlock = self._shutdown_called + if not destroy_after_unlock: self._warm_pool[sandbox_id] = (sandbox, time.time()) - destroy_after_unlock = False if destroy_after_unlock: self._destroy_quietly(sandbox, context="released during shutdown") + @staticmethod + def _output_parts(remote_path: str) -> tuple[str, ...]: + prefix = "/mnt/user-data/outputs/" + if not remote_path.startswith(prefix) or "\\" in remote_path: + raise ValueError("Output path is outside the outputs directory") + parts = remote_path[len(prefix) :].split("/") + if any(part in {"", ".", ".."} or ":" in part or part.partition(".")[0].rstrip(" ").upper() in _WINDOWS_RESERVED_NAMES or part.endswith((" ", ".")) for part in parts): + raise ValueError("Unsafe output path") + return tuple(parts) + + @staticmethod + def _output_name_key(name: str) -> str: + return unicodedata.normalize("NFC", name.casefold()) + + @classmethod + def _ambiguous_output_paths(cls, remote_paths: list[str | None]) -> set[str]: + # Component IDs keep this preflight linear in the bounded listing size, + # even for deeply nested names. A directory alias rejects all children. + components: dict[tuple[int, str], tuple[str, int]] = {} + ambiguous: set[int] = set() + path_nodes: dict[str, list[int]] = {} + for remote_path in remote_paths: + if remote_path is None: + continue + try: + parts = cls._output_parts(remote_path) + except ValueError: + continue + parent = 0 + nodes = path_nodes.setdefault(remote_path, []) + for part in parts: + key = (parent, cls._output_name_key(part)) + spelling, node = components.setdefault(key, (part, len(components) + 1)) + if spelling != part: + ambiguous.add(node) + nodes.append(node) + parent = node + return {path for path, nodes in path_nodes.items() if ambiguous.intersection(nodes)} + + @classmethod + def _output_destination(cls, root: Path, remote_path: str) -> Path: + parts = cls._output_parts(remote_path) + target = root.joinpath(*PurePosixPath(*parts).parts) + # Reject links/junctions in existing parents, including the owner/thread + # bucket. Resolving the root first would trust an escaped linked root. + for component in reversed((target, *target.parents)): + if component.is_symlink() or component.is_junction(): + raise ValueError("Symlinked output destination") + parent = root + for part in parts: + if parent.is_dir(): + key = cls._output_name_key(part) + with os.scandir(parent) as entries: + if any(entry.name != part and cls._output_name_key(entry.name) == key for entry in entries): + raise ValueError("Output destination aliases an existing path") + parent = parent / part + target.parent.mkdir(parents=True, exist_ok=True) + if target.exists() and not stat.S_ISREG(target.stat().st_mode): + raise ValueError("Output destination is not a regular file") + return target + + def _sync_outputs_to_host(self, sandbox: OpenSandboxSandbox, *, thread_id: str, user_id: str) -> None: + """Best-effort bounded mirror before parking a remote sandbox.""" + from deerflow.config.paths import get_paths + + root = get_paths().sandbox_outputs_dir(thread_id, user_id=user_id) + deadline = time.monotonic() + self._SYNC_DEADLINE_SECONDS + remote_root = shlex.quote("/mnt/user-data/outputs") + command = ( + # POSIX sh lacks pipefail: preserve each stage's own exit status. + 'set +e; ( for _tool in find head base64 mktemp; do command -v "$_tool" >/dev/null || exit 127; done; ' + f"[ -d {remote_root} ] && [ ! -L {remote_root} ] || exit 1; " + "_st=$(mktemp -d) || exit 1; " + f'{{ find {remote_root} -type f -print0; echo $? > "$_st/find"; }} | ' + f'{{ head -c {self._MAX_SYNC_LISTING_BYTES + 1}; echo $? > "$_st/head"; }} | base64; _enc=$?; ' + '_find=$(cat "$_st/find" 2>/dev/null); _head=$(cat "$_st/head" 2>/dev/null); ' + 'rm -f "$_st/find" "$_st/head"; rmdir "$_st"; ' + 'printf "\\n__DF_OUTPUT_SYNC__:%s:%s:%s\\n" "$_find" "$_head" "$_enc"; exit 0 )' + ) + execution = sandbox._run(command, timeout=self._SYNC_DEADLINE_SECONDS) + if getattr(execution, "exit_code", None) != 0 or getattr(execution, "error", None) is not None: + raise OSError("OpenSandbox output listing failed") + # SSE stdout is line-oriented. Base64 keeps event boundaries from + # inserting newlines into filenames (NUL frames alone cannot do that). + payload, marker = execution_stdout(execution).rsplit("\n", 1) + if marker not in {"__DF_OUTPUT_SYNC__:0:0:0", "__DF_OUTPUT_SYNC__:1:0:0", "__DF_OUTPUT_SYNC__:141:0:0"}: + raise OSError(f"OpenSandbox output listing failed: {marker}") + raw_listing = base64.b64decode("".join(payload.split()), validate=True) + if marker == "__DF_OUTPUT_SYNC__:141:0:0" and len(raw_listing) <= self._MAX_SYNC_LISTING_BYTES: + raise OSError("OpenSandbox output listing failed: unexpected SIGPIPE") + if len(raw_listing) > self._MAX_SYNC_LISTING_BYTES: + logger.warning("OpenSandbox output listing exceeded %d bytes; skipping synchronization", self._MAX_SYNC_LISTING_BYTES) + return + if raw_listing and not raw_listing.endswith(b"\0"): + raise OSError("OpenSandbox output listing was incomplete") + if marker == "__DF_OUTPUT_SYNC__:1:0:0": + logger.warning("OpenSandbox output listing had traversal errors; synchronizing complete records") + remote_paths: list[str | None] = [] + for remote_record in filter(None, raw_listing.split(b"\0")): + try: + remote_paths.append(remote_record.decode("utf-8")) + except UnicodeDecodeError: + logger.warning("Skipping non-UTF-8 OpenSandbox output %r", remote_record) + remote_paths.append(None) + # Include records beyond the transfer cap: publishing the first of two + # aliases would otherwise expose its bytes through both artifact URLs. + ambiguous_paths = self._ambiguous_output_paths(remote_paths) + total = 0 + for index, remote_path in enumerate(remote_paths): + if index >= self._MAX_SYNC_FILES or total >= self._MAX_SYNC_TOTAL_BYTES or time.monotonic() >= deadline: + logger.warning("OpenSandbox output synchronization stopped at its file, byte or time limit") + break + if remote_path is None: + continue + if remote_path in ambiguous_paths: + logger.warning("Skipping ambiguous OpenSandbox output %r", remote_path) + continue + staged: Path | None = None + transferring = False + try: + target = self._output_destination(root, remote_path) + transferring = True + data = sandbox._download_file_bounded(remote_path, max_bytes=min(_MAX_DOWNLOAD_SIZE, self._MAX_SYNC_TOTAL_BYTES - total), deadline=deadline) + transferring = False + total += len(data) + with tempfile.NamedTemporaryFile(prefix=".opensandbox-sync-", dir=target.parent, delete=False) as output: + staged = Path(output.name) + output.write(data) + # Recheck confinement after the network operation before + # publishing, preserving any previous complete version on error. + self._output_destination(root, remote_path) + staged.replace(target) + except (OSError, ValueError): + logger.warning("Could not synchronize OpenSandbox output %r", remote_path, exc_info=True) + # A failed stream may already have consumed bytes that the + # caller cannot account for. Stop rather than reset its budget. + if transferring: + break + finally: + if staged is not None: + staged.unlink(missing_ok=True) + def reset(self) -> None: """Park active clients so the detached provider still owns their cleanup.""" with self._lock: diff --git a/backend/packages/harness/deerflow/community/opensandbox/sandbox.py b/backend/packages/harness/deerflow/community/opensandbox/sandbox.py index 6b9517c0c3..275d8059e3 100644 --- a/backend/packages/harness/deerflow/community/opensandbox/sandbox.py +++ b/backend/packages/harness/deerflow/community/opensandbox/sandbox.py @@ -7,6 +7,8 @@ import posixpath import re import shlex import threading +import time +from contextlib import suppress from datetime import timedelta from typing import TYPE_CHECKING, Any @@ -299,11 +301,30 @@ class OpenSandboxSandbox(Sandbox): self._file_op(lambda files: files.write_file(resolved, content, mode=644)) def download_file(self, path: str, *, max_bytes: int | None = None) -> bytes: - limit = download_limit(max_bytes, _MAX_DOWNLOAD_SIZE) + return self._download_file_bounded(path, max_bytes=download_limit(max_bytes, _MAX_DOWNLOAD_SIZE)) + + def _download_file_bounded(self, path: str, *, max_bytes: int, deadline: float | None = None) -> bytes: + """Use the native streaming transport with the release pass's remaining budget.""" resolved = self._resolve_download_path(path) def read_bounded(files) -> bytes: - return read_download_stream(files.read_bytes_stream(resolved), limit, path) + stream = files.read_bytes_stream(resolved) + if deadline is None: + return read_download_stream(stream, max_bytes, path) + + def timed_chunks(): + try: + for chunk in stream: + if time.monotonic() >= deadline: + raise TimeoutError("OpenSandbox output synchronization timed out") + yield chunk + finally: + close = getattr(stream, "close", None) + if callable(close): + with suppress(Exception): + close() + + return read_download_stream(timed_chunks(), max_bytes, path) try: return self._file_op(read_bounded) diff --git a/backend/packages/harness/deerflow/sandbox/acquire_serialization.py b/backend/packages/harness/deerflow/sandbox/acquire_serialization.py index 86fb6a98d9..15d6c07999 100644 --- a/backend/packages/harness/deerflow/sandbox/acquire_serialization.py +++ b/backend/packages/harness/deerflow/sandbox/acquire_serialization.py @@ -26,6 +26,10 @@ from functools import partial DEFAULT_MAX_WORKERS = min(32, (os.cpu_count() or 1) + 4) +class AcquireSerializerClosed(RuntimeError): + """A scope checkout or queued acquire was rejected during shutdown.""" + + @dataclass class _Entry: lock: threading.Lock = field(default_factory=threading.Lock) @@ -154,7 +158,7 @@ class AcquireSerializer[KeyT: Hashable]: worker_future = self._executor.submit(acquire.run) except RuntimeError: # executor shut down between checkout and scheduling self._checkin(key, entry) - raise RuntimeError("AcquireSerializer is closed") from None + raise AcquireSerializerClosed("AcquireSerializer is closed") from None # concurrent.futures callbacks run in the cancelling/worker thread, so # executor-shutdown cleanup does not depend on this loop progressing. worker_future.add_done_callback(acquire.worker_done) @@ -172,7 +176,7 @@ class AcquireSerializer[KeyT: Hashable]: # only pending cancellation (no stale count left by an outer # swallowed CancelledError). acquire.cancel_queued() - raise RuntimeError("AcquireSerializer is closed") from None + raise AcquireSerializerClosed("AcquireSerializer is closed") from None if acquire_future.cancelled(): acquire.cancel_queued() else: @@ -204,7 +208,7 @@ class AcquireSerializer[KeyT: Hashable]: def _checkout(self, key: KeyT) -> _Entry: with self._table_lock: if self._closed: - raise RuntimeError("AcquireSerializer is closed") + raise AcquireSerializerClosed("AcquireSerializer is closed") entry = self._table.get(key) if entry is None: entry = _Entry() diff --git a/backend/tests/test_acquire_serialization.py b/backend/tests/test_acquire_serialization.py index 273e42e086..23edfae430 100644 --- a/backend/tests/test_acquire_serialization.py +++ b/backend/tests/test_acquire_serialization.py @@ -9,7 +9,7 @@ from concurrent.futures import ThreadPoolExecutor import pytest -from deerflow.sandbox.acquire_serialization import AcquireSerializer +from deerflow.sandbox.acquire_serialization import AcquireSerializer, AcquireSerializerClosed from deerflow.trace_context import get_current_trace_id, request_trace_context @@ -209,9 +209,10 @@ class TestClose: serializer = AcquireSerializer() serializer.close() serializer.close() - with pytest.raises(RuntimeError, match="closed"): + with pytest.raises(RuntimeError, match="closed") as error: with serializer.hold("k"): pass + assert isinstance(error.value, AcquireSerializerClosed) @pytest.mark.asyncio async def test_close_does_not_invalidate_active_critical_section(self): @@ -219,17 +220,19 @@ class TestClose: async with serializer.hold_async("k"): serializer.close() # must not raise, must not break exit # New holders rejected afterwards: - with pytest.raises(RuntimeError, match="closed"): + with pytest.raises(RuntimeError, match="closed") as error: async with serializer.hold_async("k"): pass + assert isinstance(error.value, AcquireSerializerClosed) @pytest.mark.asyncio async def test_executor_shutdown_between_checkout_and_schedule_reclaims_entry(self): serializer = AcquireSerializer() serializer._executor.shutdown(wait=False, cancel_futures=True) # simulates close() landing mid-window - with pytest.raises(RuntimeError, match="closed"): + with pytest.raises(RuntimeError, match="closed") as error: async with serializer.hold_async("k"): pass + assert isinstance(error.value, AcquireSerializerClosed) assert len(serializer._table) == 0 @pytest.mark.asyncio @@ -245,8 +248,9 @@ class TestClose: waiter = asyncio.create_task(_hold_once(serializer, "k")) await asyncio.sleep(0.1) # let the waiter check out and queue its acquire serializer.close() - with pytest.raises(RuntimeError, match="closed"): + with pytest.raises(RuntimeError, match="closed") as error: await waiter + assert isinstance(error.value, AcquireSerializerClosed) blocker.set() await occupied assert len(serializer._table) == 0 diff --git a/backend/tests/test_agent_guidance_check.py b/backend/tests/test_agent_guidance_check.py index 7ec83d9950..a97f8b536a 100644 --- a/backend/tests/test_agent_guidance_check.py +++ b/backend/tests/test_agent_guidance_check.py @@ -28,6 +28,7 @@ EXPECTED_GUIDANCE_PATHS = { "backend/packages/harness/deerflow/community/tavily/AGENTS.md", "backend/packages/harness/deerflow/community/e2b_sandbox/AGENTS.md", "backend/packages/harness/deerflow/community/aio_sandbox/AGENTS.md", + "backend/packages/harness/deerflow/community/opensandbox/AGENTS.md", "backend/packages/harness/deerflow/config/AGENTS.md", "backend/packages/harness/deerflow/extensions/AGENTS.md", "backend/packages/harness/deerflow/runtime/AGENTS.md", diff --git a/backend/tests/test_opensandbox_provider.py b/backend/tests/test_opensandbox_provider.py index 2fc555392e..7809dab9a3 100644 --- a/backend/tests/test_opensandbox_provider.py +++ b/backend/tests/test_opensandbox_provider.py @@ -9,6 +9,7 @@ transport, search parsing, path guards, and terminal-session eviction. from __future__ import annotations import asyncio +import base64 import errno import logging import os @@ -22,12 +23,14 @@ import time import types from dataclasses import dataclass, field from datetime import timedelta +from pathlib import Path from typing import Any import pytest from deerflow.community.opensandbox.provider import OpenSandboxProvider, _import_sdk from deerflow.community.opensandbox.sandbox import OpenSandboxSandbox +from deerflow.config.paths import Paths @dataclass @@ -139,6 +142,9 @@ class _FakeCommands: return _execution(exit_code=9) if command == "missing-complete": return _execution(stderr=("stream ended",), exit_code=None) + if "__DF_OUTPUT_SYNC__" in command: + listing = "\0".join(self._owner.file_data) + "\0" if self._owner.file_data else "" + return _execution(stdout=(base64.b64encode(listing.encode()).decode(), "__DF_OUTPUT_SYNC__:0:0:0")) if "__DF_SEARCH_STATUS__:" in command: return self._search(command) if command.startswith("find ") or "find -H " in command: @@ -467,6 +473,578 @@ def test_release_and_same_scope_warm_reclaim(monkeypatch: pytest.MonkeyPatch) -> provider.shutdown() +def test_release_mirrors_outputs_for_gateway_download(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + from app.gateway.routers.artifacts import _read_artifact_payload + + paths = Paths(base_dir=tmp_path) + monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: paths) + provider, sdk = _install(monkeypatch) + try: + sandbox_id = provider.acquire("thread", user_id="owner") + box = provider.get(sandbox_id) + box.write_file("/mnt/user-data/outputs/report.txt", "generated report") + box.update_file("/mnt/user-data/outputs/charts/result.png", bytes(range(256))) + provider.release(sandbox_id) + + outputs = paths.sandbox_outputs_dir("thread", user_id="owner") + assert _read_artifact_payload(outputs / "report.txt", "report.txt", download=True)[0] == "file" + assert (outputs / "report.txt").read_text(encoding="utf-8") == "generated report" + assert (outputs / "charts/result.png").read_bytes() == bytes(range(256)) + assert not paths.sandbox_outputs_dir("thread", user_id="other-owner").exists() + assert provider.acquire("thread", user_id="owner") == sandbox_id + assert len(sdk.remotes) == 1 + finally: + provider.shutdown() + + +@pytest.fixture +def output_sync_provider(monkeypatch, tmp_path): + paths = Paths(base_dir=tmp_path) + monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: paths) + provider, sdk = _install(monkeypatch) + sandbox_id = provider.acquire("thread", user_id="owner") + yield provider, sdk.remotes[0], sandbox_id, paths.sandbox_outputs_dir("thread", user_id="owner") + provider.shutdown() + + +def test_output_sync_refreshes_existing_content_and_keeps_literal_filename(output_sync_provider): + provider, remote, sandbox_id, outputs = output_sync_provider + virtual = "/mnt/user-data/outputs/report # 中文.txt" + remote.file_data[virtual] = b"first version" + provider.release(sandbox_id) + assert (outputs / "report # 中文.txt").read_bytes() == b"first version" + provider.acquire("thread", user_id="owner") + remote.file_data[virtual] = b"newer version" + provider.release(sandbox_id) + assert (outputs / "report # 中文.txt").read_bytes() == b"newer version" + assert not list(outputs.glob(".opensandbox-sync-*")) + + +@pytest.mark.parametrize("file_limit", [1, 2]) +def test_output_sync_skips_invalid_utf8_record_within_file_budget(output_sync_provider, monkeypatch, caplog, file_limit): + provider, remote, sandbox_id, outputs = output_sync_provider + remote.file_data["/mnt/user-data/outputs/report.txt"] = b"valid report" + listing = b"/mnt/user-data/outputs/\xff\0/mnt/user-data/outputs/report.txt\0" + monkeypatch.setattr(provider, "_MAX_SYNC_FILES", file_limit) + monkeypatch.setattr(remote.commands, "run", lambda *args, **kwargs: _execution(stdout=(base64.b64encode(listing).decode(), "__DF_OUTPUT_SYNC__:0:0:0"))) + provider.release(sandbox_id) + if file_limit == 2: + assert (outputs / "report.txt").read_bytes() == b"valid report" + else: + assert not outputs.exists() + assert remote.files.calls == [] + assert "non-UTF-8" in caplog.text + assert sandbox_id in provider._warm_pool + + +def test_output_sync_refreshes_same_size_edits(output_sync_provider): + provider, remote, sandbox_id, outputs = output_sync_provider + remote.file_data["/mnt/user-data/outputs/report.txt"] = b"first" + provider.release(sandbox_id) + assert (outputs / "report.txt").read_bytes() == b"first" + provider.acquire("thread", user_id="owner") + remote.file_data["/mnt/user-data/outputs/report.txt"] = b"later" + provider.release(sandbox_id) + assert (outputs / "report.txt").read_bytes() == b"later" + + +@pytest.mark.parametrize( + "names", + [ + ("Report.txt", "report.txt"), + ("Charts/report.txt", "charts/report.txt"), + ("Charts/first.txt", "charts/second.txt"), + ("nested/Report.txt", "nested/report.txt"), + ("Charts", "charts/report.txt"), + ("Résumé.txt", "résumé.txt"), + ("résumé.txt", "re\u0301sume\u0301.txt"), + ], +) +@pytest.mark.parametrize("reverse", [False, True]) +def test_output_sync_rejects_ambiguous_remote_names_before_copying(output_sync_provider, caplog, names, reverse): + provider, remote, sandbox_id, outputs = output_sync_provider + names = names[::-1] if reverse else names + remote.file_data.update({f"/mnt/user-data/outputs/{name}": f"content {index}".encode() for index, name in enumerate(names)}) + remote.file_data["/mnt/user-data/outputs/safe.txt"] = b"unambiguous report" + + provider.release(sandbox_id) + + assert sorted(path.relative_to(outputs).as_posix() for path in outputs.rglob("*") if path.is_file()) == ["safe.txt"] + assert (outputs / "safe.txt").read_bytes() == b"unambiguous report" + assert remote.files.calls == [("read_bytes_stream", "/mnt/user-data/outputs/safe.txt")] + assert "ambiguous" in caplog.text + assert sandbox_id in provider._warm_pool + + +def test_output_sync_collision_preflight_includes_records_beyond_file_limit(output_sync_provider, monkeypatch): + provider, remote, sandbox_id, outputs = output_sync_provider + remote.file_data.update( + { + "/mnt/user-data/outputs/Report.txt": b"first report", + "/mnt/user-data/outputs/safe.txt": b"safe report", + "/mnt/user-data/outputs/report.txt": b"different report", + } + ) + monkeypatch.setattr(provider, "_MAX_SYNC_FILES", 2) + + provider.release(sandbox_id) + + assert sorted(path.name for path in outputs.iterdir()) == ["safe.txt"] + assert remote.files.calls == [("read_bytes_stream", "/mnt/user-data/outputs/safe.txt")] + + +def test_output_sync_remote_collision_preserves_previous_copy(output_sync_provider): + provider, remote, sandbox_id, outputs = output_sync_provider + outputs.mkdir(parents=True) + (outputs / "Report.txt").write_bytes(b"previous complete report") + remote.file_data.update({"/mnt/user-data/outputs/Report.txt": b"new upper report", "/mnt/user-data/outputs/report.txt": b"new lower report"}) + + provider.release(sandbox_id) + + assert (outputs / "Report.txt").read_bytes() == b"previous complete report" + assert sorted(path.name for path in outputs.iterdir()) == ["Report.txt"] + assert remote.files.calls == [] + + +@pytest.mark.parametrize("names", [("Report.txt", "report.txt"), ("Charts/old.txt", "charts/new.txt"), ("Résumé.txt", "résumé.txt"), ("résumé.txt", "re\u0301sume\u0301.txt")]) +def test_output_sync_preserves_existing_host_alias(output_sync_provider, names): + provider, remote, sandbox_id, outputs = output_sync_provider + existing, incoming = names + previous = outputs / existing + previous.parent.mkdir(parents=True) + previous.write_bytes(b"previous complete report") + remote.file_data[f"/mnt/user-data/outputs/{incoming}"] = b"different remote report" + remote.file_data["/mnt/user-data/outputs/safe.txt"] = b"safe report" + + provider.release(sandbox_id) + + assert previous.read_bytes() == b"previous complete report" + assert sorted(path.relative_to(outputs).as_posix() for path in outputs.rglob("*") if path.is_file()) == sorted([existing, "safe.txt"]) + assert remote.files.calls == [("read_bytes_stream", "/mnt/user-data/outputs/safe.txt")] + + +def test_output_sync_rejects_alias_planted_during_download(output_sync_provider, monkeypatch): + provider, remote, sandbox_id, outputs = output_sync_provider + remote.file_data["/mnt/user-data/outputs/Report.txt"] = b"remote report" + read_stream = remote.files.read_bytes_stream + + def plant_alias(path): + (outputs / "report.txt").write_bytes(b"other complete report") + yield from read_stream(path) + + monkeypatch.setattr(remote.files, "read_bytes_stream", plant_alias) + provider.release(sandbox_id) + + assert (outputs / "report.txt").read_bytes() == b"other complete report" + assert sorted(path.name for path in outputs.iterdir()) == ["report.txt"] + + +def test_output_sync_keeps_distinct_files_in_the_same_directory(output_sync_provider): + provider, remote, sandbox_id, outputs = output_sync_provider + remote.file_data.update({"/mnt/user-data/outputs/Charts/first.txt": b"first", "/mnt/user-data/outputs/Charts/second.txt": b"second"}) + + provider.release(sandbox_id) + + assert (outputs / "Charts/first.txt").read_bytes() == b"first" + assert (outputs / "Charts/second.txt").read_bytes() == b"second" + + +@pytest.mark.parametrize("name", ["con.txt", "PRN", "aux .log", "NUL.bin", "COM9", "lpt1.doc", "COM¹.txt", "LPT²", "CONIN$.txt", "CONOUT$"]) +def test_output_destination_rejects_reserved_components_without_deprecated_api(tmp_path, monkeypatch, name): + from pathlib import PureWindowsPath + + def unavailable(*args): + raise AssertionError("PureWindowsPath.is_reserved is unavailable") + + monkeypatch.setattr(PureWindowsPath, "is_reserved", unavailable) + with pytest.raises(ValueError, match="Unsafe"): + OpenSandboxProvider._output_destination(tmp_path / "outputs", f"/mnt/user-data/outputs/{name}") + assert not (tmp_path / "outputs").exists() + + +@pytest.mark.parametrize("name", ["CONSOLE.txt", "COM10.txt", "LPT0", "report # 中文.txt"]) +def test_output_destination_accepts_nonreserved_names_without_deprecated_api(tmp_path, monkeypatch, name): + from pathlib import PureWindowsPath + + def unavailable(*args): + raise AssertionError("PureWindowsPath.is_reserved is unavailable") + + monkeypatch.setattr(PureWindowsPath, "is_reserved", unavailable) + outputs = tmp_path / "outputs" + assert OpenSandboxProvider._output_destination(outputs, f"/mnt/user-data/outputs/{name}") == outputs / name + + +@pytest.mark.parametrize("remote_path", ["/tmp/outside", "/mnt/user-data/uploads/input.txt", "/mnt/user-data/outputs/../outside", "/mnt/user-data/outputs//bad", "/mnt/user-data/outputs/C:bad", "/mnt/user-data/outputs/NUL.txt"]) +def test_output_sync_rejects_unconfined_paths(output_sync_provider, remote_path): + provider, remote, sandbox_id, outputs = output_sync_provider + remote.file_data[remote_path] = b"untrusted listing" + provider.release(sandbox_id) + assert remote.files.calls == [] + assert not outputs.exists() + assert sandbox_id in provider._warm_pool + + +@pytest.mark.parametrize("link_root", [True, False]) +def test_output_sync_rejects_host_symlink_components(output_sync_provider, tmp_path, link_root): + from support.symlinks import symlink_or_skip + + provider, remote, sandbox_id, outputs = output_sync_provider + outside = tmp_path / "outside" + outside.mkdir() + outputs.parent.mkdir(parents=True, exist_ok=True) + if link_root: + symlink_or_skip(outputs, outside, target_is_directory=True) + name = "report.txt" + else: + outputs.mkdir() + symlink_or_skip(outputs / "nested", outside, target_is_directory=True) + name = "nested/report.txt" + remote.file_data[f"/mnt/user-data/outputs/{name}"] = b"must not escape" + provider.release(sandbox_id) + assert list(outside.iterdir()) == [] + assert remote.files.calls == [] + + +def test_output_sync_transfer_failure_preserves_previous_copy(output_sync_provider, caplog): + provider, remote, sandbox_id, outputs = output_sync_provider + outputs.mkdir(parents=True) + (outputs / "report.txt").write_bytes(b"old complete version") + remote.file_data["/mnt/user-data/outputs/report.txt"] = b"new version" + remote.file_error = RuntimeError("temporary file transport failure") + provider.release(sandbox_id) + assert (outputs / "report.txt").read_bytes() == b"old complete version" + assert sandbox_id in provider._warm_pool + assert "Could not synchronize" in caplog.text + + +def test_output_sync_failed_replace_cleans_staging(output_sync_provider, monkeypatch): + provider, remote, sandbox_id, outputs = output_sync_provider + outputs.mkdir(parents=True) + (outputs / "report.txt").write_bytes(b"old complete version") + remote.file_data["/mnt/user-data/outputs/report.txt"] = b"new version" + + def fail_replace(self, target): + raise PermissionError("destination temporarily open") + + monkeypatch.setattr(Path, "replace", fail_replace) + provider.release(sandbox_id) + assert (outputs / "report.txt").read_bytes() == b"old complete version" + assert sorted(p.name for p in outputs.iterdir()) == ["report.txt"] + + +def test_output_sync_partial_download_failure_stops_remaining_transfers(output_sync_provider, monkeypatch): + provider, remote, sandbox_id, outputs = output_sync_provider + remote.file_data.update({"/mnt/user-data/outputs/first.txt": b"123", "/mnt/user-data/outputs/second.txt": b"456"}) + started = [] + + def partial_stream(path): + started.append(path) + yield b"123" + raise OSError("connection failed after receiving bytes") + + monkeypatch.setattr(remote.files, "read_bytes_stream", partial_stream) + provider.release(sandbox_id) + assert started == ["/mnt/user-data/outputs/first.txt"] + assert not (outputs / "first.txt").exists() + assert not list(outputs.glob(".opensandbox-sync-*")) + assert sandbox_id in provider._warm_pool + + +@pytest.mark.parametrize("limit", ["files", "bytes", "listing", "deadline"]) +def test_output_sync_is_bounded(output_sync_provider, monkeypatch, caplog, limit): + provider, remote, sandbox_id, outputs = output_sync_provider + remote.file_data.update({"/mnt/user-data/outputs/first.txt": b"123", "/mnt/user-data/outputs/second.txt": b"456"}) + if limit == "files": + monkeypatch.setattr(provider, "_MAX_SYNC_FILES", 1) + elif limit == "bytes": + monkeypatch.setattr(provider, "_MAX_SYNC_TOTAL_BYTES", 4) + elif limit == "listing": + monkeypatch.setattr(provider, "_MAX_SYNC_LISTING_BYTES", 4) + else: + monkeypatch.setattr(provider, "_SYNC_DEADLINE_SECONDS", 1) + monkeypatch.setattr("deerflow.community.opensandbox.provider.time", types.SimpleNamespace(monotonic=iter([0, 2]).__next__, time=time.time)) + provider.release(sandbox_id) + if limit in {"files", "bytes"}: + assert (outputs / "first.txt").read_bytes() == b"123" + assert not (outputs / "second.txt").exists() + else: + assert not outputs.exists() + assert sandbox_id in provider._warm_pool + assert caplog.records + + +def test_output_sync_closes_stream_when_single_file_exceeds_cap(output_sync_provider, monkeypatch): + provider, remote, sandbox_id, outputs = output_sync_provider + remote.file_data["/mnt/user-data/outputs/large.txt"] = b"123456789" + monkeypatch.setattr("deerflow.community.opensandbox.provider._MAX_DOWNLOAD_SIZE", 4) + provider.release(sandbox_id) + assert not (outputs / "large.txt").exists() + assert remote.stream_closed + + +@pytest.mark.parametrize("ticks", [[2], [0, 2]]) +def test_bounded_download_deadline_closes_native_stream(monkeypatch, ticks): + remote = _FakeRemote("remote") + path = "/mnt/user-data/outputs/report.txt" + remote.file_data[path] = b"123456789" + box = _box(remote) + monkeypatch.setattr("deerflow.community.opensandbox.sandbox.time", types.SimpleNamespace(monotonic=iter(ticks).__next__)) + with pytest.raises(TimeoutError, match="synchronization timed out"): + box._download_file_bounded(path, max_bytes=100, deadline=1) + assert remote.stream_closed + + +@pytest.fixture +def output_sync_shell(output_sync_provider, monkeypatch, tmp_path): + """Run the production listing pipeline; only rebase the transport paths.""" + provider, remote, sandbox_id, outputs = output_sync_provider + if os.name == "nt": + shell = Path(os.environ.get("ProgramFiles", "C:/Program Files")) / "Git/bin/bash.exe" + if not shell.is_file(): + pytest.skip("Git Bash required for real output-listing execution on Windows") + shell = str(shell) + + def posix_path(path): + return subprocess.check_output([shell, "-c", 'cygpath -u "$1"', "sh", str(path)], text=True, encoding="utf-8").strip() + else: + shell = shutil.which("sh") + if shell is None: + pytest.skip("POSIX sh required") + posix_path = str + native = tmp_path / "remote-outputs" + native.mkdir() + native_root = posix_path(native) + fake_bin = tmp_path / "sync-bin" + fake_bin.mkdir() + + def run(command, *, opts=None): + command = command.replace("/mnt/user-data/outputs", native_root) + command = f"PATH={shlex.quote(posix_path(fake_bin))}:$PATH; " + command + proc = subprocess.run([shell, "-c", command], capture_output=True, text=True, encoding="utf-8", check=False, timeout=10) + stdout = proc.stdout + if proc.returncode == 0 and "\n__DF_OUTPUT_SYNC__:" in stdout: + payload, marker = stdout.rstrip("\n").rsplit("\n", 1) + # The host's native temporary path replaces the remote mount only + # at this SDK transport boundary; status and filename bytes survive. + try: + listing = base64.b64decode("".join(payload.split()), validate=True) + listing = listing.replace(native_root.encode(), b"/mnt/user-data/outputs") + stdout = base64.b64encode(listing).decode() + "\n" + marker + except ValueError: + pass + return _execution(stdout=tuple(stdout.split("\n")), stderr=(proc.stderr,), exit_code=proc.returncode) + + monkeypatch.setattr(remote.commands, "run", run) + return provider, remote, sandbox_id, outputs, native, fake_bin + + +def test_output_sync_real_shell_handles_empty_and_literal_filenames(output_sync_shell): + provider, remote, sandbox_id, outputs, native, _ = output_sync_shell + provider.release(sandbox_id) + assert not outputs.exists() + provider.acquire("thread", user_id="owner") + names = ["中文 # $(echo injected).txt", "nested/result.bin"] + if os.name != "nt": + names.append("newline\nand\ttab.txt") + for name in names: + payload = bytes(range(256)) + target = native / name + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(payload) + remote.file_data[f"/mnt/user-data/outputs/{name}"] = payload + provider.release(sandbox_id) + for name in names: + assert (outputs / name).read_bytes() == bytes(range(256)) + + +@pytest.mark.parametrize("binary", ["find", "head", "base64", "mktemp"]) +def test_output_sync_real_shell_propagates_each_pipeline_failure(output_sync_shell, binary, caplog): + provider, remote, sandbox_id, outputs, native, fake_bin = output_sync_shell + (native / "report.txt").write_bytes(b"report") + remote.file_data["/mnt/user-data/outputs/report.txt"] = b"report" + failing = fake_bin / binary + failing.write_text("#!/bin/sh\nexit 127\n", encoding="utf-8") + failing.chmod(0o755) + provider.release(sandbox_id) + assert not outputs.exists() + assert not remote.files.calls + assert "listing failed" in caplog.text + assert sandbox_id in provider._warm_pool + + +def test_output_sync_real_shell_listing_overflow_never_downloads(output_sync_shell, monkeypatch, caplog): + provider, remote, sandbox_id, outputs, native, _ = output_sync_shell + for index in range(30): + (native / f"file-{index}.txt").write_bytes(b"data") + monkeypatch.setattr(provider, "_MAX_SYNC_LISTING_BYTES", 16) + provider.release(sandbox_id) + assert not outputs.exists() + assert not remote.files.calls + assert "listing exceeded" in caplog.text + + +def test_output_sync_real_shell_rejects_unexpected_sigpipe(output_sync_shell, caplog): + provider, remote, sandbox_id, outputs, native, fake_bin = output_sync_shell + remote.file_data["/mnt/user-data/outputs/report.txt"] = b"report" + failing = fake_bin / "find" + failing.write_text("#!/bin/sh\nprintf '/mnt/user-data/outputs/report.txt\\0'\nexit 141\n", encoding="utf-8") + failing.chmod(0o755) + provider.release(sandbox_id) + assert not outputs.exists() and not remote.files.calls + assert "listing failed" in caplog.text + + +@pytest.mark.parametrize( + ("case", "warning"), + [ + ("complete", "traversal errors"), + ("incomplete", "listing was incomplete"), + ("oversized", "listing exceeded"), + ("head_failure", "listing failed"), + ("encoder_failure", "listing failed"), + ("find_failure", "listing failed"), + ], +) +def test_output_sync_real_shell_find_walk_error_preserves_only_complete_listing(output_sync_shell, monkeypatch, caplog, case, warning): + provider, remote, sandbox_id, outputs, native, fake_bin = output_sync_shell + remote.file_data["/mnt/user-data/outputs/report.txt"] = b"report" + find = fake_bin / "find" + tail = "printf 'unfinished'\n" if case == "incomplete" else "" + status = 2 if case == "find_failure" else 1 + find.write_text(f"#!/bin/sh\nprintf '/mnt/user-data/outputs/report.txt\\0'\n{tail}echo 'find: entry vanished' >&2\nexit {status}\n", encoding="utf-8") + find.chmod(0o755) + if case == "oversized": + monkeypatch.setattr(provider, "_MAX_SYNC_LISTING_BYTES", 16) + elif case in {"head_failure", "encoder_failure"}: + binary = "head" if case == "head_failure" else "base64" + command = "cat" if binary == "head" else "command -p base64" + failing = fake_bin / binary + failing.write_text(f"#!/bin/sh\n{command}\nexit 1\n", encoding="utf-8") + failing.chmod(0o755) + provider.release(sandbox_id) + if case == "complete": + assert (outputs / "report.txt").read_bytes() == b"report" + else: + assert not outputs.exists() and not remote.files.calls + assert warning in caplog.text + assert sandbox_id in provider._warm_pool + + +def test_output_sync_real_shell_does_not_follow_remote_links(output_sync_shell, tmp_path, caplog): + from support.symlinks import symlink_or_skip + + provider, remote, sandbox_id, outputs, native, _ = output_sync_shell + outside = tmp_path / "remote-outside" + outside.mkdir() + (outside / "secret.txt").write_bytes(b"outside") + symlink_or_skip(native / "linked", outside, target_is_directory=True) + provider.release(sandbox_id) + assert not outputs.exists() and not remote.files.calls + provider.acquire("thread", user_id="owner") + (native / "linked").unlink() + native.rmdir() + symlink_or_skip(native, outside, target_is_directory=True) + provider.release(sandbox_id) + assert not outputs.exists() and not remote.files.calls + assert "listing failed" in caplog.text + + +def test_reacquire_waits_until_release_output_sync_finishes(output_sync_provider, monkeypatch): + provider, remote, sandbox_id, outputs = output_sync_provider + remote.file_data["/mnt/user-data/outputs/report.txt"] = b"report" + box = provider.get(sandbox_id) + original = box._download_file_bounded + started, unblock, acquire_started, acquired = (threading.Event() for _ in range(4)) + errors = [] + + def blocked_download(*args, **kwargs): + started.set() + assert unblock.wait(10) + return original(*args, **kwargs) + + original_checkout = provider._acquire_serializer._checkout + + def checked_checkout(key): + result = original_checkout(key) + if threading.current_thread().name == "next-turn": + acquire_started.set() + return result + + def reacquire(): + try: + assert provider.acquire("thread", user_id="owner") == sandbox_id + acquired.set() + except BaseException as exc: + errors.append(exc) + + monkeypatch.setattr(box, "_download_file_bounded", blocked_download) + monkeypatch.setattr(provider._acquire_serializer, "_checkout", checked_checkout) + release = threading.Thread(target=provider.release, args=(sandbox_id,)) + acquire = threading.Thread(target=reacquire, name="next-turn") + try: + release.start() + assert started.wait(10) + assert sandbox_id not in provider._warm_pool + acquire.start() + assert acquire_started.wait(10) + assert not acquired.is_set() + finally: + unblock.set() + release.join(10) + if acquire.ident is not None: + acquire.join(10) + assert not release.is_alive() and not acquire.is_alive() + assert not errors and acquired.is_set() + assert (outputs / "report.txt").read_bytes() == b"report" + + +@pytest.mark.parametrize("transition", ["shutdown", "reset"]) +def test_release_is_noop_when_lifecycle_closes_before_serializer_hold(output_sync_provider, monkeypatch, transition): + provider, remote, sandbox_id, outputs = output_sync_provider + original_hold = provider._acquire_serializer.hold + + def close_before_hold(key): + getattr(provider, transition)() + return original_hold(key) + + monkeypatch.setattr(provider._acquire_serializer, "hold", close_before_hold) + provider.release(sandbox_id) + assert provider.get(sandbox_id) is None + assert not outputs.exists() + + +@pytest.mark.parametrize("untrack", [True, False]) +def test_release_tolerates_typed_closure_independent_of_message(output_sync_provider, monkeypatch, untrack): + from deerflow.sandbox import acquire_serialization + + provider, remote, sandbox_id, outputs = output_sync_provider + + def close_before_hold(key): + if untrack: + provider.reset() + raise acquire_serialization.AcquireSerializerClosed("No further scope holders accepted") + + monkeypatch.setattr(provider._acquire_serializer, "hold", close_before_hold) + if untrack: + provider.release(sandbox_id) + assert provider.get(sandbox_id) is None + else: + with pytest.raises(acquire_serialization.AcquireSerializerClosed): + provider.release(sandbox_id) + + +def test_release_does_not_swallow_unrelated_runtime_error_with_old_closure_message(output_sync_provider, monkeypatch): + provider, remote, sandbox_id, outputs = output_sync_provider + + def fail_hold(key): + provider.reset() + raise RuntimeError("AcquireSerializer is closed") + + monkeypatch.setattr(provider._acquire_serializer, "hold", fail_hold) + with pytest.raises(RuntimeError, match="AcquireSerializer is closed"): + provider.release(sandbox_id) + + @pytest.mark.parametrize("exit_code", [1, None]) def test_unhealthy_warm_entry_is_destroyed_and_replaced(monkeypatch: pytest.MonkeyPatch, exit_code: int | None) -> None: provider, sdk = _install(monkeypatch)