mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
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 <willem.jiang@gmail.com>
This commit is contained in:
Lucien
GitHub
parent
b346b17fce
commit
db46250d85
@ -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.
|
||||
|
||||
@ -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.
|
||||
@ -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.
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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",
|
||||
|
||||
@ -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)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user