feat(sandbox): add bounded AIO prewarming and lifecycle benchmarks (#6998)

* fix(benchmark): validate sandbox lifecycle and concurrent measurements

* feat(sandbox): prewarm likely AIO tool turns

(cherry picked from commit 3a1017ed032ed9331987bc3dc61f23dcdac4e905)

* fix(sandbox): bind bounded prewarming to execution leases

* docs: trim frontend guidance below its existing size limit

* fix(benchmark): separate failed lifecycle stage timings

* test(sandbox): measure prewarm overlap and unused resources through agent graphs

* fix(sandbox): finish lease and sampler cleanup after cancellation or errors

* docs(benchmark): distinguish initialization wait from model overlap

* docs(helm): align the configuration example with version 61

* fix(sandbox): preserve synchronous agent graphs when prewarming is enabled

* test(cargo): isolate native fixtures from inherited make jobserver flags

* fix(sandbox): require managed cleanup before prewarming

* fix(sandbox): log denied prewarming as an expected skip

* fix(benchmark): filter provider summary input records

---------

Co-authored-by: nothingyuancando <1579035512@qq.com>
This commit is contained in:
Totoro 2026-10-18 08:10:56 +08:00 committed by GitHub
parent 002bcc521a
commit 16d36ccd21
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
29 changed files with 3304 additions and 254 deletions

View File

@ -820,6 +820,10 @@ that no port is available instead of attempting an out-of-range bind.
See the [Sandbox Configuration Guide](backend/docs/CONFIGURATION.md#sandbox) to configure your preferred mode.
Managed async AIO runs can opt into bounded sandbox prewarming to overlap startup
with the first model call. Direct SDK graphs keep lazy acquisition. It defaults off; see the guide and
[sandbox benchmarks](backend/scripts/benchmark/sandbox/README.md) before enabling it on a shared host.
Remote directory listings report traversal failures (for example, unreadable
directories) as incomplete results, even when no entries were returned. A
missing start path is reported separately as “Directory not found.”

View File

@ -1327,6 +1327,8 @@ sandbox:
```yaml
sandbox:
use: deerflow.community.aio_sandbox:AioSandboxProvider # Docker-based sandbox
prewarm: true # optional: overlap likely sandbox work with the first model call
prewarm_max_concurrent: 1 # pending or ready-but-unused prewarms per process/provider
```
For AIO images on the supported semver line (`1.9.3` through the recommended
@ -1348,6 +1350,32 @@ is attempted, and the container is recycled instead of being returned to the
warm pool. Session-level cleanup does not clear that quarantine because a
timed-out create may commit after cleanup has already returned.
`prewarm` is disabled by default and applies only to asynchronous AIO runs with
an unconditional outer execution-lease cleanup: the Gateway/embedded run worker
and native subagent executor. Direct `create_deerflow_agent(...).ainvoke()` and
`.astream()` calls retain ordinary lazy acquisition even when this setting is on.
For likely sandbox turns (for example, code fences or explicit shell commands),
startup can overlap the first model call. Authorization is checked before
speculation and checked again when a tool uses the sandbox. The acquisition
belongs to the run's execution lease; cancellation, model failure, and turns
without a sandbox tool drain startup and release that owner. Unused sandboxes
follow the provider's normal warm-pool/idle cleanup policy, so false positives
can still consume resources until eviction.
Internal prewarm handles and the non-serializable cleanup marker belong only in
server-owned runtime context, never checkpoint state. Register the marker only
inside an unconditional async execution-lease cleanup; revoke it during cleanup,
which must drain startup and release the owner even if no sandbox ID was published.
`prewarm_max_concurrent` is a positive integer, default `1`. It bounds pending
and ready-but-unused prewarms for each provider instance in a process. When
full, DeerFlow skips speculation without queuing or delaying foreground tools.
The limit is not a total-container cap or a deployment-wide quota; multiple
Gateway processes have separate budgets. Other providers and synchronous
embedded calls keep their existing acquisition behavior. Restart after changing
these settings. Use the [sandbox benchmarks](../scripts/benchmark/sandbox/README.md)
to measure first-tool wait, unused speculation, and resource pressure on your host.
**BoxLite micro-VM Sandbox** (runs sandbox code in daemonless OCI micro-VMs):
```yaml
sandbox:

View File

@ -217,6 +217,15 @@ class SandboxConfig(BaseModel):
default=False,
description="Allow the bash tool to execute directly on the host when using LocalSandboxProvider. Dangerous; intended only for fully trusted local environments.",
)
prewarm: bool = Field(
default=False,
description=("Start an AioSandboxProvider acquire in parallel with the first model call when the user input contains a strong sandbox-use signal. Disabled by default; ignored by other sandbox providers."),
)
prewarm_max_concurrent: int = Field(
default=1,
ge=1,
description=("Maximum pending or ready-but-unused AIO prewarms per provider instance in this process. When full, skip speculation without delaying foreground tools. This is not a total container or deployment-wide limit."),
)
image: str | None = Field(
default=None,
description="Sandbox image to use (Docker/AIO, BoxLite OCI, or OpenSandbox image)",
@ -257,6 +266,7 @@ class SandboxConfig(BaseModel):
)
@field_validator(
"prewarm_max_concurrent",
"port",
"replicas",
"acquire_timeout",

View File

@ -95,7 +95,7 @@ from deerflow.runtime.serialization import serialize
from deerflow.runtime.stream_bridge import StreamBridge
from deerflow.runtime.stream_modes import normalize_stream_modes, to_langgraph_stream_modes
from deerflow.runtime.user_context import get_current_user, get_effective_user_id, resolve_runtime_user_id
from deerflow.sandbox.lease import SANDBOX_SERVER_OWNED_CONTEXT_KEYS
from deerflow.sandbox.lease import SANDBOX_SERVER_OWNED_CONTEXT_KEYS, register_sandbox_prewarm_cleanup
from deerflow.scheduler.runtime import SCHEDULER_CAPABILITY_CONTEXT_KEY, SchedulerRunCapability
from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, ensure_trace_id
from deerflow.tracing import inject_langfuse_metadata
@ -1215,6 +1215,7 @@ async def run_agent(
thread_incarnation=thread_incarnation,
scheduler_capability=ctx.scheduler_capability,
)
register_sandbox_prewarm_cleanup(runtime_ctx)
# Bind every checkpoint produced by this run to the effective agent
# identity that produced its state. Manual compaction uses only this
# server-overwritten value for memory policy; request metadata cannot

View File

@ -123,3 +123,5 @@ symlinks, and visible depth/size bounds.
- `read_file` - Read file contents with optional line range; full reads keep stored line endings (Local `read_file`/`grep` read `newline="\n"`, writes `newline=""`), ranged reads join with `\n`; remote providers slice with `split_file_lines`, not `splitlines()`
- `write_file` - Write/append to files, creates directories; overwrites by default and exposes the `append` argument in the model-facing schema for end-of-file writes; subject to the read-before-write gate when `read_before_write.enabled` (see Middleware Chain)
- `str_replace` - Substring replacement (single or all occurrences); `_match_line_endings` gives `old_str`/`new_str` a CRLF file's endings; same-path serialization is scoped to `(sandbox.id, path)` so isolated sandboxes do not contend on identical virtual paths inside one process; subject to the read-before-write gate when `read_before_write.enabled` (see Middleware Chain)
**Optional AIO prewarm**: default off, managed async AIO only. Require the server-owned cleanup marker; direct SDK stays lazy. Preserve authorization, execution-owner cleanup, runtime-only handles, permit limits and normal eviction under the [prewarm lifecycle rules](../../../../docs/CONFIGURATION.md#sandbox).

View File

@ -27,11 +27,16 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
SANDBOX_PREWARM_CONTEXT_KEY = "__deerflow_sandbox_prewarm"
SANDBOX_PREWARM_CLEANUP_CONTEXT_KEY = "__deerflow_sandbox_prewarm_cleanup"
_PREWARM_CLEANUP_MARKER = object()
SANDBOX_LEASE_OWNER_CONTEXT_KEY = "sandbox_lease_owner_id"
SANDBOX_COMMAND_SCOPE_CONTEXT_KEY = "sandbox_command_scope_id"
SANDBOX_SERVER_OWNED_CONTEXT_KEYS = frozenset(
{
SANDBOX_LEASE_OWNER_CONTEXT_KEY,
SANDBOX_PREWARM_CONTEXT_KEY,
SANDBOX_PREWARM_CLEANUP_CONTEXT_KEY,
SANDBOX_COMMAND_SCOPE_CONTEXT_KEY,
}
)
@ -734,6 +739,16 @@ def discard_sandbox_lease_manager(provider: SandboxProvider) -> None:
manager.close()
def register_sandbox_prewarm_cleanup(context: dict[str, Any]) -> None:
"""Admit prewarm only inside an unconditional async lease-release finally."""
context[SANDBOX_PREWARM_CLEANUP_CONTEXT_KEY] = _PREWARM_CLEANUP_MARKER
def sandbox_prewarm_cleanup_registered(context: Any) -> bool:
"""Do not accept a serialized caller flag as an execution cleanup boundary."""
return isinstance(context, dict) and context.get(SANDBOX_PREWARM_CLEANUP_CONTEXT_KEY) is _PREWARM_CLEANUP_MARKER
def ensure_sandbox_lease_owner(context: Any) -> str | None:
"""Create one ephemeral owner id in a mutable runtime context."""
if not isinstance(context, dict):
@ -789,13 +804,22 @@ def release_sandbox_execution_lease(context: Any) -> None:
async def release_sandbox_execution_lease_async(context: Any) -> None:
"""Async counterpart to :func:`release_sandbox_execution_lease`."""
lease = _sandbox_execution_lease(context)
if lease is None:
return
"""Drain speculative work before releasing the ordinary execution lease."""
from deerflow.sandbox.prewarm import close_sandbox_prewarm_context
from deerflow.sandbox.sandbox_provider import get_sandbox_provider
if isinstance(context, dict):
context.pop(SANDBOX_PREWARM_CLEANUP_CONTEXT_KEY, None)
try:
await close_sandbox_prewarm_context(context)
finally:
# Disabled/skipped speculation may own no lease while a later real tool
# does. Its cleanup can re-raise cancellation after draining, or fail;
# neither may bypass that ordinary execution's release. Do not return
# from this finally: cancellation must survive even without a binding.
lease = _sandbox_execution_lease(context)
if lease is not None:
from deerflow.sandbox.sandbox_provider import get_sandbox_provider
owner_id, _sandbox_id = lease
provider = get_sandbox_provider()
await get_sandbox_lease_manager(provider).release_async(owner_id)
owner_id, _sandbox_id = lease
provider = get_sandbox_provider()
await get_sandbox_lease_manager(provider).release_async(owner_id)

View File

@ -31,6 +31,7 @@ from deerflow.sandbox.lease import (
sandbox_lease_owner,
)
from deerflow.sandbox.overwrite import unwrap_sandbox
from deerflow.sandbox.prewarm import close_sandbox_prewarm, start_sandbox_prewarm
from deerflow.sandbox.sandbox_provider import get_initialized_sandbox_provider
from deerflow.utils.file_io import await_drained
@ -428,6 +429,22 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
return {"sandbox": Overwrite({"sandbox_id": retained_id})}
return await super().abefore_agent(state, runtime)
@override
def before_model(self, state: SandboxMiddlewareState, runtime: Runtime) -> dict | None:
# LangChain registers both halves of this hook as one graph node. An
# async-only override would make compiled invoke()/stream() fail even
# with prewarm disabled. Sync runs retain ordinary lazy acquisition.
return super().before_model(state, runtime)
@override
async def abefore_model(self, state: SandboxMiddlewareState, runtime: Runtime) -> dict | None:
# This deliberately starts but does not await acquisition. The model's
# first response then overlaps Docker startup; the first sandbox tool
# claims the same task through ensure_sandbox_initialized_async().
if self._lazy_init and state.get("sandbox") is None:
start_sandbox_prewarm(state, runtime)
return await super().abefore_model(state, runtime)
@override
def after_agent(self, state: SandboxMiddlewareState, runtime: Runtime) -> dict | None:
sandbox, fork_restored = unwrap_sandbox(state.get("sandbox"))
@ -463,6 +480,7 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
@override
async def aafter_agent(self, state: SandboxMiddlewareState, runtime: Runtime) -> dict | None:
await close_sandbox_prewarm(runtime)
sandbox, fork_restored = unwrap_sandbox(state.get("sandbox"))
if sandbox is not None:
sandbox_id = sandbox["sandbox_id"]

View File

@ -0,0 +1,232 @@
"""Optional AIO startup owned by the same lease as its agent execution."""
from __future__ import annotations
import asyncio
import logging
import re
import threading
from dataclasses import dataclass, field
from typing import Any
from langchain_core.messages import HumanMessage
from deerflow.authz.sandbox_authz import authorize_sandbox_execution_async, safe_app_config_async
from deerflow.config import get_app_config
from deerflow.runtime.user_context import resolve_runtime_user_id
from deerflow.sandbox.acquire_serialization import AcquireSerializerClosed
from deerflow.sandbox.exceptions import SandboxAuthorizationError
from deerflow.sandbox.lease import (
SANDBOX_PREWARM_CONTEXT_KEY,
SandboxLeaseManager,
get_sandbox_lease_manager,
run_sync_lifecycle_operation,
sandbox_lease_owner,
sandbox_prewarm_cleanup_registered,
)
from deerflow.sandbox.sandbox_provider import SandboxProvider, get_initialized_sandbox_provider, get_sandbox_provider
from deerflow.utils.file_io import await_drained
logger = logging.getLogger(__name__)
_CODE_FENCE_RE = re.compile(r"```")
_URL_RE = re.compile(r"\b[a-zA-Z][a-zA-Z0-9+.-]*://[^\s<>]+")
_PATH_RE = re.compile(r"(?:^|[\s`\"'(])(?:~/(?:[\w.-]+/)*[\w.-]+|/(?:[\w.-]+)(?:/[\w.-]+)*|[A-Za-z]:\\)")
_COMMAND_RE = re.compile(
r"^\s*(?:\$\s+\S|python(?:3)?\s+(?:-[cmV]\b|\S+\.py\b)|"
r"(?:pip|pip3)\s+(?:install|list|show|freeze|check)\b|"
r"git\s+(?:status|diff|log|show|clone|checkout|switch|fetch|pull|push|commit|add|branch)\b|"
r"(?:npm|pnpm)\s+(?:run|install|test|build|exec|add)\b|"
r"uv\s+(?:run|sync|pip|add|lock|venv)\b|"
r"pytest(?:\s+(?:-|\S+[/\\.]|tests\b)|\s*$)|make\s+(?:test|check|build|install|dev)\b|"
r"docker\s+(?:run|build|compose|ps|logs|exec|inspect)\b|kubectl\s+(?:get|describe|logs|exec|apply)\b)",
re.MULTILINE,
)
_TRACEBACK_RE = re.compile(r"(?:Traceback \(most recent call last\)|^\s*File [\"'].+[\"'], line \d+)", re.MULTILINE)
# Identity keyed and bounded by live speculative handles. A lock, rather than an
# asyncio semaphore, also covers the isolated subagent loop. No provider IO runs
# under it. Real tool acquisitions never queue behind this optional budget.
_permit_lock = threading.Lock()
_permits: dict[int, tuple[SandboxProvider, int]] = {}
def _take_permit(provider: SandboxProvider, limit: int) -> bool:
with _permit_lock:
entry = _permits.get(id(provider))
count = entry[1] if entry is not None and entry[0] is provider else 0
if count >= limit:
return False
_permits[id(provider)] = provider, count + 1
return True
def _message_text(message: HumanMessage) -> str:
content = message.content
if isinstance(content, str):
return content
if not isinstance(content, list):
return ""
return "\n".join(block["text"] for block in content if isinstance(block, dict) and isinstance(block.get("text"), str))
def should_prewarm_sandbox(state: dict[str, Any]) -> bool:
"""Predict only explicit code/command/path signals in the latest user turn."""
messages = state.get("messages")
if not isinstance(messages, list):
return False
message = next((item for item in reversed(messages) if isinstance(item, HumanMessage)), None)
if message is None:
return False
text = _URL_RE.sub("", _message_text(message))
return bool(_CODE_FENCE_RE.search(text) or _PATH_RE.search(text) or _COMMAND_RE.search(text) or _TRACEBACK_RE.search(text))
@dataclass
class SandboxPrewarm:
owner_id: str
thread_id: str
user_id: str
context: dict[str, Any]
task: asyncio.Task[str | None] = field(init=False)
provider: SandboxProvider | None = None
manager: SandboxLeaseManager | None = None
claimed: bool = False
closed: bool = False
_permit: bool = False
_cleanup_task: asyncio.Task[None] | None = None
def _return_permit(self) -> None:
if self.provider is None:
return
with _permit_lock:
if not self._permit:
return
self._permit = False
entry = _permits.get(id(self.provider))
if entry is not None and entry[0] is self.provider:
if entry[1] == 1:
del _permits[id(self.provider)]
else:
_permits[id(self.provider)] = self.provider, entry[1] - 1
async def acquire(self) -> str | None:
ready = False
try:
config = self.context.get("app_config")
if config is None:
config = await asyncio.to_thread(get_app_config)
sandbox = config.sandbox
if not getattr(sandbox, "prewarm", False) or not sandbox.use.endswith(":AioSandboxProvider"):
return None
# Startup policy is run-scoped, but authorization must be fresh;
# claiming later does not replace the ordinary tool's fresh gate.
await authorize_sandbox_execution_async(context=self.context, app_config=await safe_app_config_async())
self.provider = get_initialized_sandbox_provider()
if self.provider is None:
# Only AIO opts in here; its constructor performs filesystem and
# backend setup, so keep that work off the model's event loop.
self.provider = await run_sync_lifecycle_operation(get_sandbox_provider)
self.manager = get_sandbox_lease_manager(self.provider)
limit = getattr(sandbox, "prewarm_max_concurrent", 1)
if not _take_permit(self.provider, limit):
logger.debug("Sandbox prewarm skipped: speculative budget occupied")
return None
self._permit = True
sandbox_id = await self.manager.acquire_async(self.owner_id, self.thread_id, user_id=self.user_id)
ready = True
return sandbox_id
except SandboxAuthorizationError:
logger.debug("Sandbox prewarm skipped: sandbox execution denied")
return None
except Exception:
logger.info("Sandbox prewarm unavailable; normal tool acquisition remains authoritative", exc_info=True)
return None
finally:
if not ready:
self._return_permit()
async def claim(self, provider: SandboxProvider) -> str | None:
if self.closed:
return None
sandbox_id = await asyncio.shield(self.task)
if self.closed or sandbox_id is None:
return None
if self.provider is not provider:
await self.close()
return None
self.claimed = True
self._return_permit()
return sandbox_id
async def _cleanup(self) -> None:
try:
if not self.task.done():
self.task.cancel()
try:
await self.task
except asyncio.CancelledError:
pass # The manager reconciles any provider result before this.
finally:
if self.manager is not None:
try:
await self.manager.release_async(self.owner_id)
except AcquireSerializerClosed:
if get_initialized_sandbox_provider() is self.provider:
raise
# Provider reset/shutdown already detached its manager.
# That lifecycle owns its remaining clients; raw release
# here could close a sibling still using the old provider.
logger.debug("Prewarm provider detached; its lifecycle owns final cleanup")
finally:
self._return_permit()
# A closed marker prevents repeated speculation in goal continuations
# without retaining the authorization/config snapshot indefinitely.
self.context.clear()
async def close(self) -> None:
self.closed = True
if self._cleanup_task is None:
self._cleanup_task = asyncio.create_task(self._cleanup(), name=f"sandbox-prewarm-cleanup:{self.owner_id}")
# after_agent and the execution finally may both arrive here. Keep the
# same task reachable through repeated cancellation and reentrant cleanup.
await await_drained(self._wait_for_cleanup())
async def _wait_for_cleanup(self) -> None:
assert self._cleanup_task is not None
await asyncio.shield(self._cleanup_task)
def start_sandbox_prewarm(state: dict[str, Any], runtime: Any) -> None:
"""Start at most one optional acquisition within an admitted async run."""
context = getattr(runtime, "context", None)
if not sandbox_prewarm_cleanup_registered(context) or SANDBOX_PREWARM_CONTEXT_KEY in context or state.get("sandbox") is not None:
return
owner_id = sandbox_lease_owner(context)
thread_id = context.get("thread_id")
if owner_id is None or not isinstance(thread_id, str) or not thread_id or not should_prewarm_sandbox(state):
return
snapshot = dict(context)
handle = SandboxPrewarm(owner_id, thread_id, resolve_runtime_user_id(runtime), snapshot)
handle.task = asyncio.create_task(handle.acquire(), name=f"sandbox-prewarm:{owner_id}")
context[SANDBOX_PREWARM_CONTEXT_KEY] = handle
async def claim_sandbox_prewarm(runtime: Any, *, provider: SandboxProvider) -> str | None:
context = getattr(runtime, "context", None)
handle = context.get(SANDBOX_PREWARM_CONTEXT_KEY) if isinstance(context, dict) else None
if not isinstance(handle, SandboxPrewarm):
return None
if handle.owner_id != sandbox_lease_owner(context) or handle.thread_id != context.get("thread_id") or handle.user_id != resolve_runtime_user_id(runtime):
return None
return await handle.claim(provider)
async def close_sandbox_prewarm_context(context: Any) -> None:
handle = context.get(SANDBOX_PREWARM_CONTEXT_KEY) if isinstance(context, dict) else None
if isinstance(handle, SandboxPrewarm):
await handle.close()
async def close_sandbox_prewarm(runtime: Any) -> None:
await close_sandbox_prewarm_context(getattr(runtime, "context", None))

View File

@ -41,6 +41,7 @@ from deerflow.sandbox.lease import (
)
from deerflow.sandbox.overwrite import unwrap_sandbox
from deerflow.sandbox.path_patterns import build_output_mask_pattern, normalize_mask_tail, replace_output_path_matches
from deerflow.sandbox.prewarm import claim_sandbox_prewarm
from deerflow.sandbox.read_file_contract import (
READ_FILE_EMPTY,
READ_FILE_INVALID_END_LINE,
@ -1928,14 +1929,16 @@ async def ensure_sandbox_initialized_async(runtime: Runtime | None = None) -> Sa
provider = get_sandbox_provider()
user_id = resolve_runtime_user_id(runtime)
owner_id = sandbox_lease_owner(runtime.context)
if owner_id is None:
sandbox_id = await provider.acquire_async(thread_id, user_id=user_id)
else:
sandbox_id = await get_sandbox_lease_manager(provider).acquire_async(
owner_id,
thread_id,
user_id=user_id,
)
sandbox_id = await claim_sandbox_prewarm(runtime, provider=provider)
if sandbox_id is None:
if owner_id is None:
sandbox_id = await provider.acquire_async(thread_id, user_id=user_id)
else:
sandbox_id = await get_sandbox_lease_manager(provider).acquire_async(
owner_id,
thread_id,
user_id=user_id,
)
runtime.state["sandbox"] = {"sandbox_id": sandbox_id}

View File

@ -1823,6 +1823,9 @@ class SubagentExecutor:
context[_SANDBOX_LEASE_OWNER_CONTEXT_KEY] = sandbox_lease_owner_id
context[_SANDBOX_COMMAND_SCOPE_CONTEXT_KEY] = sandbox_lease_owner_id
execution_context = context
from deerflow.sandbox.lease import register_sandbox_prewarm_cleanup
register_sandbox_prewarm_cleanup(context)
context["agent_id"] = self.config.name
if self.loop_detection_recorder is not None:
context[LOOP_DETECTION_RECORDER_CONTEXT_KEY] = self.loop_detection_recorder
@ -2055,13 +2058,11 @@ class SubagentExecutor:
# usage before the outer wrapper chooses CANCELLED or TIMED_OUT.
if collector is not None:
result.update_token_usage_records(collector.snapshot_records())
if execution_context is not None and execution_context.get("sandbox_id") is not None:
if execution_context is not None:
try:
from deerflow.sandbox import get_sandbox_provider
from deerflow.sandbox.lease import get_sandbox_lease_manager
from deerflow.sandbox.lease import release_sandbox_execution_lease_async
provider = get_sandbox_provider()
await get_sandbox_lease_manager(provider).release_async(sandbox_lease_owner_id)
await release_sandbox_execution_lease_async(execution_context)
except Exception:
logger.warning(
"[trace=%s] Failed to release sandbox execution lease for subagent %s",

View File

@ -0,0 +1,115 @@
# Sandbox benchmarks
Run from `backend/` with a readable, isolated DeerFlow config and runtime home.
These commands create real sandbox resources; the ordinary unit suite is offline.
Pin the sandbox image and record the Git revision, Docker/VM capacity and config.
Use a disposable home and never point a benchmark at a production provider.
```sh
DEER_FLOW_CONFIG_PATH=/absolute/path/benchmark.yaml \
DEER_FLOW_HOME=/absolute/path/disposable-home \
uv run python scripts/benchmark/sandbox/bench_provider.py \
--provider aio-docker --image <pinned-image> \
--scenario cold_unique_thread --no-warmpool --iterations 10 \
--output cold.jsonl
uv run python scripts/benchmark/sandbox/bench_provider.py \
--provider aio-docker --image <same-pinned-image> \
--scenario replica_pressure --users 4 --concurrency 4 --iterations 8 \
--output pressure.jsonl
uv run python scripts/benchmark/sandbox/summarize.py cold.jsonl pressure.jsonl
```
`summarize.py` summarizes only `bench_provider.py` results. It reports and ignores
metadata and typed prewarm/resource records before aggregation and exit-status
checks. Inputs containing no provider results fail with a format diagnostic;
failed provider rows still make the summary exit nonzero.
Each invocation gets an independent run/thread namespace and AIO container prefix.
`--run-id` makes that namespace explicit. AIO cleanup uses the actual acquired ID
and the provider's destruction path, checking that destruction completed. Fault
injection stops that same resource while retaining the stale provider entry to
exercise recovery. Failures in acquisition, command execution, cleanup, or warm-up
make the command fail; inspect row counts and semantic results as well as timing.
`replica_pressure` holds simultaneous active clients behind barriers before running
commands. `barrier_wait_ms` is test synchronization overhead, so its `total_ms` is
not a user-facing latency measurement. Turn timing starts inside the worker after
it acquires its scope lock: executor-queue and scope-lock waits are excluded.
Pressure batches are submitted sequentially; later batches do not represent
requests already waiting in an admission queue. The provider's `replicas` is a soft cap;
protecting active work can exceed it. `--users` distributes independent scopes
across synthetic users, not real accounts.
Warm-up resources are destroyed before timed cases. A fresh-thread cold run does
not measure image download or a cold Docker daemon. `acquire_ms`, `run_ms`,
`release_ms` and `cleanup_ms` separate those costs; no-warm-pool `total_ms` includes
explicit destruction. `warm_hit` comes from the actual reclaim path. Keep different
images, pool settings, replica limits, user counts and concurrency levels separate;
the summary's default grouping does so. Percentiles use inclusive interpolation.
## Resource samples
Run the sampler beside an explicit benchmark run ID:
```sh
uv run python -m scripts.benchmark.sandbox.resources \
--prefix deerflow-bench-my-run- --output resources.jsonl --duration 60
# In another terminal, use --run-id my-run in the provider benchmark.
```
Only containers with the exact name prefix are sampled. This sampler never
creates or destroys resources. Failed or incomplete readings stay unknown, not
zero. Peaks are observed sample peaks; Docker memory and CPU omit host/VM overhead.
Join these records with the provider's active/warm snapshots before interpreting
resource use. Record host/daemon capacity separately and do not compare stripped
code-only images with a browser-enabled image as equivalent functionality.
## First-tool overlap and unused prewarming
The prewarm driver uses a different JSONL format; do not pass its output to
`summarize.py`. Its exit status checks case outcomes and batch cleanup. Inspect
`prewarm_case` and `batch_cleanup` records for their separate `success` fields;
the `configuration` record has no outcome and must not count as a failed case.
The provider microbenchmark does not measure first-tool overlap. For that, use
the compiled-agent experiment with the same isolated config/home and pinned image:
```sh
uv run python -m scripts.benchmark.sandbox.bench_prewarm \
--image <pinned-image> --cases tool plain fenced_no_tool model_error cancel \
--model-delays 0 1 3 --iterations 5 --output prewarm.jsonl
uv run python -m scripts.benchmark.sandbox.bench_prewarm \
--image <same-pinned-image> --cases tool fenced_no_tool cancel \
--model-delays 3 --iterations 2 --users 4 --concurrency 4 \
--replicas 2 --prewarm-max-concurrent 1 --output prewarm-pressure.jsonl
```
This uses a scripted model with production thread/sandbox middleware, the bash
tool, authorization, and execution-lease cleanup. Containers and commands are
real; model delays and users are synthetic. No paid model is called, and this is
not a standalone SDK prewarm example: the driver explicitly registers its
unconditional execution cleanup, as the Gateway/embedded worker and native
subagent executor do. Direct SDK graphs retain lazy acquisition. This is
not a Gateway HTTP/SSE or live-model latency measurement. Each A/B pair has the
same prompt and predetermined tool intent, randomized variant order, fresh thread
IDs, and exact-resource destruction between batches. Output paths must be new.
`first_tool_wait_ms` spans the sandbox initialization entry until it is usable;
the bash tool's preceding authorization scope and graph dispatch are outside
that interval. Derive response-to-ready wait from `events_ms.tool_ready -
events_ms.model_first_response` to include them. `elapsed_ms` also includes graph
construction and terminal cleanup. A reused prewarm is not necessarily model
overlap: compare acquire/model event timestamps, especially in the zero-delay
control, and do not attribute startup jitter to prewarming.
`unused_active_ms` ends when an unused acquisition is parked:
it excludes subsequent warm-pool residency. Lease cleanup does not imply physical
container destruction; ordinary idle eviction still applies. `replicas` and the
speculative budget do not enforce deployment-wide admission or user fairness.
Join per-case monotonic windows and lifecycle snapshots with the companion
`.resources.jsonl`; a whole-file peak mixes enabled/disabled cases and teardown.
Keep those phases separate when comparing memory or CPU. Record dirty-tree/driver
hashes and inspect semantic success, cleanup records, and process exit status.

View File

@ -0,0 +1,382 @@
"""Opt-in AIO prewarm A/B through a real agent graph and deterministic model.
Run from backend with an explicit isolated DEER_FLOW_CONFIG_PATH/DEER_FLOW_HOME:
python -m scripts.benchmark.sandbox.bench_prewarm --output prewarm.jsonl
No paid model calls. The graph uses production thread/sandbox middleware, bash
tool, authorization and execution-lease cleanup. This measures that bounded
agent path, not HTTP/SSE latency or live-model behavior. Cases start with fresh
thread identities; exact acquired resources are destroyed between batches.
"""
from __future__ import annotations
import argparse
import asyncio
import contextvars
import hashlib
import json
import math
import random
import subprocess
import time
import uuid
from contextlib import contextmanager
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from langchain.agents import create_agent
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from pydantic import Field
from deerflow.agents.middlewares.thread_data_middleware import ThreadDataMiddleware
from deerflow.agents.thread_state import ThreadState
from deerflow.config import get_app_config
from deerflow.sandbox import tools as sandbox_tools
from deerflow.sandbox.lease import get_sandbox_lease_manager, register_sandbox_prewarm_cleanup, release_sandbox_execution_lease_async, sandbox_lease_owner
from deerflow.sandbox.middleware import SandboxMiddleware
from deerflow.sandbox.sandbox_provider import set_sandbox_provider, shutdown_sandbox_provider
from scripts.benchmark.sandbox.bench_provider import _evict_from_warm, _make_aio_provider, _pool_counts
from scripts.benchmark.sandbox.resources import DockerResourceSampler
COMMAND_MARKER = "deerflow-bench-command-ok"
COMMAND = f"python -c \"print('{COMMAND_MARKER}')\""
PROMPTS = {
"tool": f"Run this command:\n```sh\n{COMMAND}\n```",
"plain": "Explain why the sky is blue. Do not run any tools.",
"fenced_no_tool": "Explain this example without running it:\n```python\nprint(42)\n```",
"model_error": "Run this example:\n```python\nprint(42)\n```",
"cancel": f"Run this command:\n```sh\n{COMMAND}\n```",
}
@dataclass
class Trace:
started: float = field(default_factory=time.monotonic)
timestamps: dict[str, float] = field(default_factory=dict)
acquisitions: list[dict[str, Any]] = field(default_factory=list)
releases: list[str] = field(default_factory=list)
sandbox_ids: set[str] = field(default_factory=set)
model_started: asyncio.Event = field(default_factory=asyncio.Event)
acquire_started: asyncio.Event = field(default_factory=asyncio.Event)
acquire_ready: asyncio.Event = field(default_factory=asyncio.Event)
tool_requested: asyncio.Event = field(default_factory=asyncio.Event)
cancel_requested: asyncio.Event = field(default_factory=asyncio.Event)
tool_sandbox_id: str | None = None
cleanup_error_type: str | None = None
lifecycle_samples: list[dict[str, Any]] = field(default_factory=list)
def mark(self, name: str) -> None:
self.timestamps.setdefault(name, time.monotonic())
event = getattr(self, name, None)
if isinstance(event, asyncio.Event):
event.set()
_TRACE: contextvars.ContextVar[Trace | None] = contextvars.ContextVar("benchmark_trace", default=None)
@contextmanager
def observe_provider(provider):
"""Observation-only wrappers in this standalone process; restore on exit."""
acquire, release, initialize = provider.acquire_async, provider.release, sandbox_tools.ensure_sandbox_initialized_async
def sample(trace, event):
active, warm = _pool_counts(provider)
trace.lifecycle_samples.append({"event": event, "monotonic_s": time.monotonic(), "active": active, "warm": warm})
async def timed_acquire(*args, **kwargs):
trace = _TRACE.get()
if trace is None:
return await acquire(*args, **kwargs)
trace.mark("acquire_started")
entry = {"started": time.monotonic(), "sandbox_id": None}
trace.acquisitions.append(entry)
try:
sid = await acquire(*args, **kwargs)
entry.update(ready=time.monotonic(), sandbox_id=sid)
trace.sandbox_ids.add(sid)
trace.mark("acquire_ready")
sample(trace, "acquire_ready")
return sid
finally:
entry["finished"] = time.monotonic()
def timed_release(sid):
result = release(sid)
if (trace := _TRACE.get()) is not None:
trace.sandbox_ids.add(sid)
trace.releases.append(sid)
trace.timestamps["released"] = time.monotonic()
sample(trace, "released")
return result
async def timed_initialize(runtime=None):
trace = _TRACE.get()
if trace is not None:
trace.mark("tool_requested")
sandbox = await initialize(runtime)
if trace is not None:
trace.mark("tool_ready")
trace.tool_sandbox_id = sandbox.id
return sandbox
provider.acquire_async, provider.release = timed_acquire, timed_release
sandbox_tools.ensure_sandbox_initialized_async = timed_initialize
try:
yield
finally:
provider.acquire_async, provider.release = acquire, release
sandbox_tools.ensure_sandbox_initialized_async = initialize
class ScriptedModel(BaseChatModel):
"""Exactly one predetermined bash request, or a predetermined terminal case."""
trace: Any = Field(exclude=True)
case: str
delay: float
gate: Any = Field(default=None, exclude=True)
@property
def _llm_type(self):
return "offline-sandbox-benchmark"
def bind_tools(self, tools, **kwargs):
return self
def _generate(self, *args, **kwargs):
raise AssertionError("benchmark must use the asynchronous model path")
async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs):
if any(isinstance(message, ToolMessage) for message in messages):
answer = AIMessage(content="Finished the fixed command.")
else:
self.trace.mark("model_started")
if self.gate is not None:
await self.gate.wait()
await asyncio.sleep(self.delay)
self.trace.mark("model_first_response")
if self.case == "model_error":
raise RuntimeError("scripted model failure")
answer = AIMessage(content="Fixed explanation; no tools required.")
if self.case in ("tool", "cancel"):
answer = AIMessage(content="", tool_calls=[{"name": "bash", "args": {"command": COMMAND, "description": "Deterministic benchmark command"}, "id": "bench-command", "type": "tool_call"}])
return ChatResult(generations=[ChatGeneration(message=answer)])
async def run_case(provider, app_config, *, case: str, prewarm: bool, model_delay: float, thread_id: str, user_id: str, trace: Trace | None = None, model_gate=None) -> dict[str, Any]:
"""Run the real graph; even errors/cancellation use the runtime's final fence."""
if case not in PROMPTS:
raise ValueError(f"Unknown benchmark case: {case}")
trace = trace or Trace()
config = app_config.model_copy(deep=True)
config.sandbox.prewarm = prewarm
context = {"thread_id": thread_id, "user_id": user_id, "app_config": config}
model = ScriptedModel(trace=trace, case=case, delay=model_delay, gate=model_gate)
graph = create_agent(model, tools=[sandbox_tools.bash_tool], middleware=[ThreadDataMiddleware(), SandboxMiddleware()], state_schema=ThreadState)
output = None
outcome = "completed"
error = None
async def invoke():
token = _TRACE.set(trace)
try:
register_sandbox_prewarm_cleanup(context)
return await graph.ainvoke({"messages": [HumanMessage(content=PROMPTS[case])]}, context=context, config={"configurable": {"thread_id": thread_id}, "recursion_limit": 10})
finally:
trace.mark("cleanup_started")
try:
await release_sandbox_execution_lease_async(context)
except BaseException as exc:
trace.cleanup_error_type = type(exc).__name__
raise
finally:
trace.mark("cleanup_finished")
_TRACE.reset(token)
task = asyncio.create_task(invoke())
try:
if case == "cancel":
await asyncio.wait_for(trace.acquire_started.wait(), timeout=60)
trace.mark("cancel_requested")
task.cancel()
output = await task
except asyncio.CancelledError:
outcome = "cancelled"
except Exception as exc:
outcome = "model_error" if case == "model_error" and str(exc) == "scripted model failure" else "failed"
error = type(exc).__name__
finally:
if not task.done():
task.cancel()
await asyncio.gather(task, return_exceptions=True)
tool_messages = [message for message in (output or {}).get("messages", []) if isinstance(message, ToolMessage)]
tool_valid = len(tool_messages) == 1 and str(tool_messages[0].content).strip() == COMMAND_MARKER
owner = sandbox_lease_owner(context)
lease_remaining = owner is not None and get_sandbox_lease_manager(provider).binding_for(owner) is not None
requested = trace.timestamps.get("tool_requested", float("inf"))
speculative = [entry for entry in trace.acquisitions if entry["started"] < requested and entry["sandbox_id"] is not None]
def duration(start: str, end: str) -> float | None:
stamps = trace.timestamps
return round((stamps[end] - stamps[start]) * 1000, 3) if start in stamps and end in stamps else None
expected_outcome = {"cancel": "cancelled", "model_error": "model_error"}.get(case, "completed")
success = outcome == expected_outcome and not lease_remaining and trace.cleanup_error_type is None and (tool_valid if case == "tool" else not tool_messages)
active, warm = _pool_counts(provider)
finished = time.monotonic()
return {
"record_type": "prewarm_case",
"case": case,
"prewarm": prewarm,
"user_id": user_id,
"thread_id": thread_id,
"model_delay_seconds": model_delay,
"outcome": outcome,
"success": success,
"error_type": error,
"cleanup_error_type": trace.cleanup_error_type,
"lifecycle_samples": trace.lifecycle_samples,
"acquire_calls": len(trace.acquisitions),
"sandbox_ids": sorted(trace.sandbox_ids),
"release_calls": len(trace.releases),
"prewarm_reused": bool(speculative and trace.tool_sandbox_id in {entry["sandbox_id"] for entry in speculative}),
"unused_prewarm": bool(speculative and not trace.tool_requested.is_set()),
"tool_output_valid": tool_valid,
"lease_remaining": lease_remaining,
"active_after_cleanup": active,
"warm_after_cleanup": warm,
"first_tool_wait_ms": duration("tool_requested", "tool_ready"),
"cleanup_ms": duration("cleanup_started", "cleanup_finished"),
"cancel_to_cleanup_ms": duration("cancel_requested", "cleanup_finished"),
"elapsed_ms": round((finished - trace.started) * 1000, 3),
"finished_utc": datetime.now(UTC).isoformat(),
"started_monotonic_s": trace.started,
"finished_monotonic_s": finished,
"events_ms": {name: round((stamp - trace.started) * 1000, 3) for name, stamp in trace.timestamps.items()},
"acquire_elapsed_ms": round(sum(entry["finished"] - entry["started"] for entry in trace.acquisitions) * 1000, 3),
"unused_active_ms": duration("acquire_ready", "released") if speculative and not trace.tool_requested.is_set() else 0.0,
}
async def run_matrix(provider, config, args, stream) -> bool:
rng = random.Random(args.seed)
success = True
with observe_provider(provider):
for case in args.cases:
for delay in args.model_delays:
for iteration in range(args.iterations):
variants = [False, True]
rng.shuffle(variants)
for prewarm in variants:
rows = await asyncio.gather(
*[run_case(provider, config, case=case, prewarm=prewarm, model_delay=delay, user_id=f"bench-user-{lane % args.users}", thread_id=f"{args.run_id}-{uuid.uuid4().hex[:12]}") for lane in range(args.concurrency)]
)
for row in rows:
row.update(iteration=iteration, concurrency=args.concurrency, users=args.users)
stream.write(json.dumps(row, allow_nan=False) + "\n")
stream.flush()
success = success and row["success"]
# Separate measurement from reset; only destroy IDs
# returned by this batch, after all execution holders drain.
sandbox_ids = sorted({sid for row in rows for sid in row["sandbox_ids"]})
reset_started = time.monotonic()
errors = []
for sid in sandbox_ids:
try:
await asyncio.to_thread(_evict_from_warm, provider, sid)
except Exception as exc:
errors.append({"sandbox_id": sid, "error_type": type(exc).__name__})
stream.write(
json.dumps(
{
"record_type": "batch_cleanup",
"case": case,
"prewarm": prewarm,
"model_delay_seconds": delay,
"iteration": iteration,
"sandbox_ids": sandbox_ids,
"success": not errors,
"errors": errors,
"started_monotonic_s": reset_started,
"finished_monotonic_s": time.monotonic(),
}
)
+ "\n"
)
stream.flush()
if errors:
return False
return success
def main(argv=None) -> int:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("--output", type=Path, required=True)
parser.add_argument("--cases", nargs="+", choices=list(PROMPTS), default=list(PROMPTS))
parser.add_argument("--model-delays", nargs="+", type=float, default=[0, 1, 3])
parser.add_argument("--iterations", type=int, default=1)
parser.add_argument("--concurrency", type=int, default=1)
parser.add_argument("--users", type=int, default=1)
parser.add_argument("--replicas", type=int, default=3)
parser.add_argument("--prewarm-max-concurrent", type=int, default=1)
parser.add_argument("--seed", type=int, default=5002)
parser.add_argument("--image")
parser.add_argument("--resource-interval", type=float, default=1)
args = parser.parse_args(argv)
if min(args.iterations, args.concurrency, args.users, args.replicas, args.prewarm_max_concurrent) < 1 or any(not math.isfinite(delay) or delay < 0 for delay in args.model_delays):
parser.error("counts must be positive and model delays finite/nonnegative")
if not math.isfinite(args.resource_interval) or args.resource_interval <= 0:
parser.error("resource interval must be finite and positive")
args.run_id = uuid.uuid4().hex[:12]
config = get_app_config().model_copy(deep=True)
if not config.sandbox.use.endswith(":AioSandboxProvider"):
parser.error("AIO prewarm comparison requires an AioSandboxProvider config; refusing a silently disabled treatment")
config.sandbox.prewarm_max_concurrent = args.prewarm_max_concurrent
prefix = f"deerflow-bench-{args.run_id}"
revision = subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True, text=True, check=True).stdout.strip()
status = subprocess.run(["git", "status", "--porcelain"], capture_output=True, check=True).stdout
patch = subprocess.run(["git", "diff", "HEAD", "--binary"], capture_output=True, check=True).stdout
sampler = DockerResourceSampler(prefix + "-", args.output.with_suffix(".resources.jsonl"), interval=args.resource_interval)
with args.output.open("x", encoding="utf-8") as stream:
provider, effective = _make_aio_provider({"image": args.image, "container_prefix": prefix, "replicas": args.replicas, "idle_timeout": 600})
set_sandbox_provider(provider)
try:
stream.write(
json.dumps(
{
"record_type": "configuration",
"revision": revision,
"git_dirty": bool(status),
"tracked_patch_sha256": hashlib.sha256(patch).hexdigest(),
"driver_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
"seed": args.seed,
"image": effective["image"],
"container_prefix": prefix,
"prewarm_max_concurrent": args.prewarm_max_concurrent,
"replicas": args.replicas,
"model": "scripted-no-paid-api",
"matrix": {key: getattr(args, key) for key in ("cases", "model_delays", "iterations", "concurrency", "users")},
}
)
+ "\n"
)
sampler.start()
sampler.wait_for_first_sample(timeout=10)
success = asyncio.run(run_matrix(provider, config, args, stream))
finally:
try:
shutdown_sandbox_provider()
finally:
sampler.stop()
return 0 if success else 1
if __name__ == "__main__":
raise SystemExit(main())

View File

@ -28,11 +28,11 @@ Providers
Scenarios
---------
``warm_same_thread`` Reuse one ``(user_id, thread_id)`` — warm pool hit after first turn.
``warm_same_thread`` Reuse a scope per concurrent lane — warm pool hit after first turn.
``cold_unique_thread`` Fresh ``thread_id`` per turn — never hits warm pool.
``warm_miss_many_threads`` Rotate through N distinct threads — verifies isolation.
``idle_timeout`` Release, sleep > timeout, re-acquire — verify reaper works.
``replica_pressure`` Push past ``replicas`` — verify eviction only targets warm entries.
``replica_pressure`` Hold concurrent active clients past the replica soft cap.
Workloads
---------
@ -50,10 +50,13 @@ import importlib
import importlib.metadata
import json
import os
import re
import statistics
import sys
import threading
import time
import types
import uuid
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor, as_completed
from contextlib import contextmanager
@ -86,6 +89,15 @@ class BenchResult:
health_check_skip_seconds: float | None = None
image: str | None = None
no_warmpool: bool = False
sandbox_id: str | None = None
users: int = 1
run_id: str | None = None
# These are instantaneous provider snapshots, not Docker/VM resource metrics.
active_at_acquire: int | None = None
warm_at_acquire: int | None = None
cleanup_ms: float = 0.0
barrier_wait_ms: float = 0.0
fault_injected: bool = False
# ── Workloads ───────────────────────────────────────────────────────────
@ -220,6 +232,7 @@ def _make_boxlite_provider(config: dict[str, Any]) -> tuple[Any, dict[str, Any]]
def _make_aio_provider(config: dict[str, Any]) -> tuple[Any, dict[str, Any]]:
"""Create an AioSandboxProvider with stub config."""
from deerflow.community.aio_sandbox.aio_sandbox_provider import AioSandboxProvider
from deerflow.config.sandbox_config import SandboxNetworkConfig
sandbox_attrs = {
"image": config.get("image"),
@ -231,14 +244,16 @@ def _make_aio_provider(config: dict[str, Any]) -> tuple[Any, dict[str, Any]]:
"environment": config.get("environment", {}),
"provisioner_url": config.get("provisioner_url", ""),
}
stub = _stub_config(sandbox_attrs)
stub.sandbox.network = SandboxNetworkConfig()
with _patched_module_attr(
"deerflow.community.aio_sandbox.aio_sandbox_provider",
"get_app_config",
lambda: _stub_config(sandbox_attrs),
lambda: stub,
):
provider = AioSandboxProvider()
return provider, sandbox_attrs
return provider, {**sandbox_attrs, "image": provider._config["image"]}
PROVIDER_FACTORIES: dict[str, Callable] = {
@ -282,32 +297,45 @@ def _warm_hit_from_acquire() -> bool:
return bool(getattr(_WARM_HIT_STATE, "value", False))
def _compute_sandbox_id(provider: Any, thread_id: str, user_id: str) -> str:
"""Compute the deterministic sandbox_id the provider would use."""
if hasattr(provider, "_sandbox_id"):
return provider._sandbox_id(thread_id, user_id)
# Fallback: use the provider's own method or hash
import hashlib
return hashlib.sha256(f"{user_id}:{thread_id}".encode()).hexdigest()[:8]
def _was_warm_hit(provider: Any, sandbox_id: str) -> bool:
"""Check if the sandbox_id is currently in the provider's warm pool."""
with provider._lock:
return sandbox_id in provider._warm_pool
def _evict_from_warm(provider: Any, sandbox_id: str) -> None:
"""Forcibly remove and destroy a warm-pool entry (no-warmpool simulation)."""
"""Destroy exactly this acquired resource; never pop AIO metadata first."""
with provider._lock:
entry = provider._warm_pool.pop(sandbox_id, None)
if entry is not None:
box, _ = entry
try:
box.close()
except Exception:
pass
entry = provider._warm_pool.get(sandbox_id)
destroy = getattr(provider, "destroy", None)
if callable(destroy):
if destroy(sandbox_id) is False:
raise RuntimeError(f"Destroy deferred for {sandbox_id}; cold baseline is invalid")
backend = getattr(provider, "_backend", None)
if entry is not None and backend is not None and backend.is_alive(entry[0]):
raise RuntimeError(f"Container still alive after destroy: {sandbox_id}")
elif callable(getattr(provider, "_invalidate_box", None)):
# BoxLite has no public destroy; its terminal-invalidation path owns all
# identity/health-check bookkeeping and propagates close failures.
provider._invalidate_box(sandbox_id, reason="benchmark_no_warmpool")
else:
raise RuntimeError("Provider has no supported benchmark destroy adapter")
def _crash_warm_resource(provider: Any, sandbox_id: str) -> None:
"""Crash the exact returned resource, leaving its stale entry for recovery."""
with provider._lock:
entry = provider._warm_pool.get(sandbox_id)
if entry is None:
raise RuntimeError(f"No warm resource to fault: {sandbox_id}")
resource = entry[0]
backend = getattr(provider, "_backend", None)
if backend is not None:
backend.destroy(resource)
if backend.is_alive(resource):
raise RuntimeError(f"Fault injection did not stop {sandbox_id}")
else:
resource.close()
def _pool_counts(provider: Any) -> tuple[int | None, int]:
with provider._lock:
count_active = getattr(provider, "_active_count_locked", None)
return count_active() if callable(count_active) else None, len(provider._warm_pool)
# ── Core benchmark runner ───────────────────────────────────────────────
@ -326,32 +354,42 @@ def _run_one_turn(
no_warmpool: bool,
state_write_turn: bool = False,
expected_state: str | None = None,
acquired_barrier: threading.Barrier | None = None,
) -> BenchResult:
"""Execute one acquire→run→release cycle and return a BenchResult."""
t0 = time.perf_counter()
sandbox_id = _compute_sandbox_id(provider, thread_id, user_id)
sid: str | None = None
warm_hit: bool | None = None
acquire_ms = 0.0
run_ms = 0.0
release_ms = 0.0
release_needed = False
cleanup_ms = 0.0
active_count: int | None = None
warm_count: int | None = None
barrier_wait_ms = 0.0
try:
tracked_warm_hit = getattr(provider, "_bench_warm_hit_tracking_installed", False)
_reset_warm_hit_tracking()
if not tracked_warm_hit:
warm_hit = _was_warm_hit(provider, sandbox_id)
t_a = time.perf_counter()
sid = provider.acquire(thread_id, user_id=user_id)
t_b = time.perf_counter()
try:
sid = provider.acquire(thread_id, user_id=user_id)
finally:
acquire_ms = (time.perf_counter() - t_a) * 1000
if tracked_warm_hit:
warm_hit = _warm_hit_from_acquire()
acquire_ms = (t_b - t_a) * 1000
release_needed = True
if acquired_barrier is not None:
t_barrier = time.perf_counter()
acquired_barrier.wait(timeout=120)
active_count, warm_count = _pool_counts(provider)
if acquired_barrier is not None:
# Keep every acquired client alive until all snapshots are taken.
acquired_barrier.wait(timeout=120)
barrier_wait_ms = (time.perf_counter() - t_barrier) * 1000
sandbox = provider.get(sid)
if sandbox is None:
raise RuntimeError(f"acquire returned {sid!r} but get() returned None")
@ -363,10 +401,13 @@ def _run_one_turn(
cmd = _STATE_READ
t_c = time.perf_counter()
output = sandbox.execute_command(cmd, timeout=30)
t_d = time.perf_counter()
run_ms = (t_d - t_c) * 1000
if output.startswith("Error:"):
try:
output = sandbox.execute_command(cmd, timeout=30)
finally:
run_ms = (time.perf_counter() - t_c) * 1000
# Sandbox.execute_command is a text API. Both adapters render nonzero
# exits in-band; AIO also reports indeterminate/terminated sessions.
if output.startswith("Error:") or re.search(r"(?:Exit Code:|Command exited with code)\s*-?[1-9]\d*", output) or "command outcome is unknown" in output.casefold() or "command was terminated" in output.casefold():
raise RuntimeError(output)
if expected_state is not None:
@ -374,13 +415,18 @@ def _run_one_turn(
raise RuntimeError(f"State reuse failed: expected {expected_state!r} in output, got {output.strip()!r}")
t_e = time.perf_counter()
provider.release(sid)
try:
provider.release(sid)
finally:
release_ms += (time.perf_counter() - t_e) * 1000
release_needed = False
t_f = time.perf_counter()
release_ms = (t_f - t_e) * 1000
if no_warmpool:
_evict_from_warm(provider, sid)
t_cleanup = time.perf_counter()
try:
_evict_from_warm(provider, sid)
finally:
cleanup_ms += (time.perf_counter() - t_cleanup) * 1000
return BenchResult(
provider=provider_name,
@ -393,24 +439,36 @@ def _run_one_turn(
acquire_ms=acquire_ms,
run_ms=run_ms,
release_ms=release_ms,
total_ms=(t_f - t0) * 1000,
total_ms=(time.perf_counter() - t0) * 1000,
warm_hit=warm_hit,
success=True,
no_warmpool=no_warmpool,
sandbox_id=sid,
active_at_acquire=active_count,
warm_at_acquire=warm_count,
cleanup_ms=cleanup_ms,
barrier_wait_ms=barrier_wait_ms,
)
except Exception as exc:
if acquired_barrier is not None:
acquired_barrier.abort()
release_error: str | None = None
if sid is not None and release_needed:
t_release = time.perf_counter()
try:
provider.release(sid)
t_release = time.perf_counter()
try:
provider.release(sid)
finally:
release_ms += (time.perf_counter() - t_release) * 1000
if no_warmpool:
_evict_from_warm(provider, sid)
t_cleanup = time.perf_counter()
try:
_evict_from_warm(provider, sid)
finally:
cleanup_ms += (time.perf_counter() - t_cleanup) * 1000
except Exception as release_exc:
release_error = repr(release_exc)
finally:
release_ms = (time.perf_counter() - t_release) * 1000
error = str(exc) if str(exc).startswith("Error:") else repr(exc)
if release_error is not None:
error = f"{error}; release_error={release_error}"
@ -430,9 +488,25 @@ def _run_one_turn(
success=False,
error=error,
no_warmpool=no_warmpool,
sandbox_id=sid,
active_at_acquire=active_count,
warm_at_acquire=warm_count,
cleanup_ms=cleanup_ms,
barrier_wait_ms=barrier_wait_ms,
)
def _write_results(results: list[BenchResult], output_path: Path, config_used: dict[str, Any]) -> None:
for result in results:
result.replicas = config_used.get("replicas")
result.idle_timeout = config_used.get("idle_timeout")
result.health_check_skip_seconds = config_used.get("health_check_skip_seconds")
result.image = config_used.get("image")
with output_path.open("a", encoding="utf-8") as stream:
for result in results:
stream.write(json.dumps(asdict(result), ensure_ascii=False) + "\n")
def _run_scenario(
provider: Any,
provider_name: str,
@ -444,111 +518,79 @@ def _run_scenario(
no_warmpool: bool,
config_used: dict[str, Any],
fault_inject_after: int | None = None,
users: int = 1,
run_id: str | None = None,
) -> list[BenchResult]:
command = WORKLOADS.get(workload_name, WORKLOADS["noop"])
# For state_reuse workload, run paired turns: write → read
run_id = run_id or uuid.uuid4().hex[:12]
is_state_reuse = workload_name == "state_reuse"
pressure_slots = max(concurrency, int(config_used.get("replicas", 3)) * 2)
scope_locks: dict[tuple[str, str], threading.Lock] = {}
scope_lock_guard = threading.Lock()
results: list[BenchResult] = []
def _run_one(i: int) -> BenchResult:
def _run_one(i: int, barrier: threading.Barrier | None = None) -> BenchResult:
# Repeated state turns stay paired per user; every invocation gets a
# fresh namespace, so an earlier process cannot masquerade as cold.
user_index = (i // 2 if is_state_reuse else i) % users
user_id = f"bench-user-{user_index}"
if scenario == "cold_unique_thread":
tid = f"cold-{i}"
elif scenario == "warm_same_thread":
tid = "warm-hit"
elif scenario == "warm_miss_many_threads":
tid = f"thread-{i % max(concurrency, 4)}"
elif scenario in ("idle_timeout", "replica_pressure"):
tid = f"warm-hit-{i % concurrency}"
elif scenario == "replica_pressure":
tid = f"pressure-{i % pressure_slots}"
else:
tid = f"default-{i}"
state_write = is_state_reuse and (i % 2 == 0)
expect_state = "benchmark-state-42" if is_state_reuse and (i % 2 == 1) else None
return _run_one_turn(
provider=provider,
provider_name=provider_name,
scenario=scenario,
workload_name=workload_name,
command=command,
iteration=i,
concurrency=concurrency,
user_id="bench-user",
thread_id=tid,
no_warmpool=no_warmpool,
state_write_turn=state_write,
expected_state=expect_state,
)
def _tid(i: int) -> str:
if scenario == "cold_unique_thread":
return f"cold-{i}"
elif scenario == "warm_same_thread":
return "warm-hit"
elif scenario == "warm_miss_many_threads":
return f"thread-{i % max(concurrency, 4)}"
elif scenario in ("idle_timeout", "replica_pressure"):
return f"warm-hit-{i % concurrency}"
else:
return f"default-{i}"
def _inject_fault(i: int) -> None:
if fault_inject_after is None or i != fault_inject_after:
return
tid = _tid(i)
sandbox_id = _compute_sandbox_id(provider, tid, "bench-user")
with provider._lock:
warm_entry = provider._warm_pool.get(sandbox_id)
if warm_entry is not None:
box, _ = warm_entry
try:
box.close()
except Exception:
pass
print(
f" [fault] killed warm-pool box {sandbox_id} after iteration {i}",
file=sys.stderr,
)
else:
print(
f" [fault] no warm-pool box {sandbox_id} to kill after iteration {i}",
file=sys.stderr,
tid = f"thread-{i % max(concurrency, 4)}"
tid = f"{run_id}-{tid}"
with scope_lock_guard:
scope_lock = scope_locks.setdefault((user_id, tid), threading.Lock())
with scope_lock:
result = _run_one_turn(
provider,
provider_name,
scenario,
workload_name,
command,
i,
concurrency,
user_id,
tid,
no_warmpool,
state_write_turn=is_state_reuse and i % 2 == 0,
expected_state="benchmark-state-42" if is_state_reuse and i % 2 == 1 else None,
acquired_barrier=barrier,
)
result.users = users
result.run_id = run_id
if fault_inject_after == i and result.success:
try:
if result.sandbox_id is None:
raise RuntimeError("Fault injection requires a returned sandbox ID")
_crash_warm_resource(provider, result.sandbox_id)
result.fault_injected = True
except Exception as exc:
result.success = False
result.error = f"Fault injection failed: {exc!r}"
return result
results: list[BenchResult] = []
if concurrency == 1:
for i in range(iterations):
r = _run_one(i)
results.append(r)
_inject_fault(i)
results = [_run_one(i) for i in range(iterations)]
else:
sem = threading.BoundedSemaphore(concurrency)
def _guarded(i: int) -> BenchResult:
with sem:
return _run_one(i)
with ThreadPoolExecutor(max_workers=concurrency) as pool:
futures = {pool.submit(_guarded, i): i for i in range(iterations)}
for future in as_completed(futures):
i = futures[future]
results.append(future.result())
_inject_fault(i)
# Annotate with config
for r in results:
r.replicas = config_used.get("replicas")
r.idle_timeout = config_used.get("idle_timeout")
r.health_check_skip_seconds = config_used.get("health_check_skip_seconds")
r.image = config_used.get("image")
# Write JSONL
with output_path.open("a", encoding="utf-8") as f:
for r in results:
f.write(json.dumps(asdict(r), ensure_ascii=False) + "\n")
if scenario == "replica_pressure":
# Real simultaneous active clients, including the last partial
# batch. Barrier failures abort peers rather than leak holders.
for start in range(0, iterations, concurrency):
batch = range(start, min(start + concurrency, iterations))
barrier = threading.Barrier(len(batch))
futures = [pool.submit(_run_one, i, barrier) for i in batch]
results.extend(future.result() for future in as_completed(futures))
else:
futures = [pool.submit(_run_one, i) for i in range(iterations)]
results.extend(future.result() for future in as_completed(futures))
results.sort(key=lambda result: result.iteration)
_write_results(results, output_path, config_used)
return results
@ -602,6 +644,8 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
default="bench_results.jsonl",
help="JSONL output file (appended, default: bench_results.jsonl)",
)
p.add_argument("--users", type=int, default=1, help="Distinct user identities, assigned round-robin (default: 1)")
p.add_argument("--run-id", default=None, help="Thread namespace; defaults to a fresh ID to avoid reuse from an earlier invocation")
p.add_argument(
"--no-warmpool",
action="store_true",
@ -651,8 +695,17 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
def main(argv: list[str] | None = None) -> int:
args = _parse_args(argv)
if min(args.concurrency, args.users, args.replicas) < 1 or min(args.iterations, args.warmup_iterations) < 0:
raise SystemExit("concurrency, users and replicas must be positive; iteration counts must be nonnegative")
if args.fault_inject is not None and (args.concurrency != 1 or args.no_warmpool or args.scenario != "warm_same_thread"):
raise SystemExit("fault injection requires warm_same_thread, concurrency=1 and an enabled warm pool")
if args.fault_inject is not None and not 0 <= args.fault_inject < args.iterations:
raise SystemExit("fault injection iteration must be inside the measured batch")
if args.scenario == "idle_timeout" and (args.concurrency != 1 or args.idle_timeout <= 0):
raise SystemExit("idle_timeout requires concurrency=1 and a positive idle timeout")
if args.workload == "state_reuse" and (args.scenario != "warm_same_thread" or args.concurrency != 1):
raise SystemExit("state_reuse requires --scenario warm_same_thread --concurrency 1")
args.run_id = args.run_id or uuid.uuid4().hex[:12]
output_path = Path(args.output)
@ -661,6 +714,8 @@ def main(argv: list[str] | None = None) -> int:
"idle_timeout": args.idle_timeout,
"health_check_skip_seconds": args.health_check_skip_seconds,
"image": args.image,
# Never discover/clean containers belonging to a user's regular stack.
"container_prefix": f"deerflow-bench-{args.run_id}",
}
factory = PROVIDER_FACTORIES[args.provider]
@ -680,7 +735,7 @@ def main(argv: list[str] | None = None) -> int:
file=sys.stderr,
)
for i in range(args.warmup_iterations):
_run_one_turn(
warmup = _run_one_turn(
provider=provider,
provider_name=args.provider,
scenario="warmup",
@ -689,9 +744,14 @@ def main(argv: list[str] | None = None) -> int:
iteration=-(args.warmup_iterations - i),
concurrency=1,
user_id="bench-user",
thread_id="warmup",
no_warmpool=args.no_warmpool,
thread_id=f"{args.run_id}-warmup",
no_warmpool=True,
)
if not warmup.success:
warmup.run_id = args.run_id
_write_results([warmup], output_path, config_used)
print(f"Warm-up failed: {warmup.error}", file=sys.stderr)
return 1
# --- Idle-timeout scenario: special handling ---
if args.scenario == "idle_timeout":
@ -718,15 +778,16 @@ def main(argv: list[str] | None = None) -> int:
no_warmpool=args.no_warmpool,
config_used=config_used,
fault_inject_after=args.fault_inject,
users=args.users,
run_id=args.run_id,
)
_print_summary(results, args)
return int(any(not result.success for result in results))
finally:
provider.shutdown()
return 0
def _run_idle_timeout_scenario(
provider: Any,
@ -740,7 +801,7 @@ def _run_idle_timeout_scenario(
benchmark. We call ``_reap_expired_warm`` directly after the sleep to
simulate the reaper firing.
"""
idle = min(args.idle_timeout, 10)
idle = args.idle_timeout
print(
f"Idle timeout scenario: acquire, release, force-reap after {idle + 1}s sleep (timeout={idle}s)",
file=sys.stderr,
@ -748,7 +809,8 @@ def _run_idle_timeout_scenario(
results: list[BenchResult] = []
for i in range(min(args.iterations, 20)):
tid = f"idle-{i}"
tid = f"{args.run_id}-idle-{i}"
user_id = f"bench-user-{i % args.users}"
r1 = _run_one_turn(
provider,
args.provider,
@ -757,7 +819,7 @@ def _run_idle_timeout_scenario(
WORKLOADS.get(args.workload, "true"),
i * 2,
args.concurrency,
"bench-user",
user_id,
tid,
args.no_warmpool,
)
@ -776,11 +838,13 @@ def _run_idle_timeout_scenario(
WORKLOADS.get(args.workload, "true"),
i * 2 + 1,
args.concurrency,
"bench-user",
user_id,
tid,
args.no_warmpool,
)
if r2.warm_hit:
r2.success = False
r2.error = "Idle-timeout reaping retained a warm entry"
print(
f" WARNING: turn {i * 2 + 1} was a warm hit — reaping may not have removed the entry",
file=sys.stderr,
@ -788,16 +852,12 @@ def _run_idle_timeout_scenario(
results.append(r2)
for r in results:
r.replicas = config_used.get("replicas")
r.idle_timeout = config_used.get("idle_timeout")
r.image = config_used.get("image")
with output_path.open("a", encoding="utf-8") as f:
for r in results:
f.write(json.dumps(asdict(r), ensure_ascii=False) + "\n")
r.users = args.users
r.run_id = args.run_id
_write_results(results, output_path, config_used)
_print_summary(results, args)
return 0
return int(any(not result.success for result in results))
def _run_replica_pressure_scenario(
@ -806,53 +866,22 @@ def _run_replica_pressure_scenario(
output_path: Path,
config_used: dict[str, Any],
) -> int:
"""Push past replicas limit to verify eviction behaviour."""
replicas = args.replicas
overcommit = replicas * 2
print(
f"Replica pressure: replicas={replicas}, overcommitting to {overcommit} unique threads, {args.iterations} rounds",
file=sys.stderr,
"""Hold real concurrent clients; replicas is a soft warm-pool cap."""
results = _run_scenario(
provider,
args.provider,
"replica_pressure",
args.workload,
args.iterations,
args.concurrency,
output_path,
args.no_warmpool,
config_used,
users=args.users,
run_id=args.run_id,
)
results: list[BenchResult] = []
for i in range(args.iterations):
tid = f"pressure-{i % overcommit}"
r = _run_one_turn(
provider,
args.provider,
"replica_pressure",
args.workload,
WORKLOADS.get(args.workload, "true"),
i,
args.concurrency,
"bench-user",
tid,
args.no_warmpool,
)
# Track warm pool evictions
with provider._lock:
warm_size = len(provider._warm_pool)
active_size = len(provider._boxes)
print(
f" iter {i}: warm_pool={warm_size} active={active_size} warm_hit={r.warm_hit}",
file=sys.stderr,
)
results.append(r)
for r in results:
r.replicas = config_used.get("replicas")
r.idle_timeout = config_used.get("idle_timeout")
r.image = config_used.get("image")
with output_path.open("a", encoding="utf-8") as f:
for r in results:
f.write(json.dumps(asdict(r), ensure_ascii=False) + "\n")
_print_summary(results, args)
return 0
return int(any(not result.success for result in results))
def _print_summary(results: list[BenchResult], args: argparse.Namespace) -> None:
@ -868,11 +897,11 @@ def _print_summary(results: list[BenchResult], args: argparse.Namespace) -> None
print(f" {r.error}", file=sys.stderr)
return
def _p(arr: list[float], pct: float) -> float:
if not arr:
return 0.0
idx = max(0, min(len(arr) - 1, int(len(arr) * pct / 100)))
return sorted(arr)[idx]
def _p(arr: list[float], pct: int) -> float:
if len(arr) < 2:
return arr[0] if arr else 0.0
# Match summarize.py: inclusive, linearly interpolated percentiles.
return statistics.quantiles(arr, n=100, method="inclusive")[pct - 1]
a = [r.acquire_ms for r in ok]
t = [r.total_ms for r in ok]

View File

@ -0,0 +1,268 @@
#!/usr/bin/env python3
"""Sample resources for a uniquely prefixed Docker benchmark run, without mutation.
Run from ``backend`` alongside the benchmark::
python -m scripts.benchmark.sandbox.resources \
--prefix deerflow-bench-<run-id>- --output resources.jsonl --duration 60
The caller must reserve a unique prefix and use it for benchmark containers.
Only matching *running* containers are measured. An empty successful inventory
means zero; unavailable, partial or invalid measurements remain unknown (null).
Peaks are peaks of samples, not guaranteed process maxima. Docker's reported
memory excludes cache on Linux; CPU is Docker's percentage and may exceed 100.
This does not measure Docker VM overhead or assign active/warm lifecycle states.
Use the provider benchmark's lifecycle records alongside these samples.
"""
from __future__ import annotations
import argparse
import json
import math
import re
import subprocess
import threading
import time
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, TextIO
def _validate_prefix(prefix: str) -> None:
if not isinstance(prefix, str) or len(prefix) < 8 or not re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_.-]*-", prefix):
raise ValueError("Use a unique container prefix of at least 8 characters ending in '-'; only Docker name characters are allowed")
def parse_bytes(value: str) -> int | None:
"""Parse Docker's decimal SI and binary IEC memory values; invalid is unknown."""
match = re.fullmatch(r"\s*(\d+(?:\.\d+)?)\s*([kKmMgGtTpP]?[iI]?[bB])\s*", value)
if match is None:
return None
number = float(match.group(1))
unit = match.group(2).lower()
power = "bkmgtp".index(unit[0]) if unit != "b" else 0
amount = number * (1024 if "i" in unit else 1000) ** power
return int(amount) if math.isfinite(amount) else None
def _cpu_percent(value: str) -> float | None:
if not re.fullmatch(r"\d+(?:\.\d+)?%", value.strip()):
return None
result = float(value.strip()[:-1])
return result if math.isfinite(result) else None
def _docker(command: list[str], timeout: float) -> tuple[list[dict[str, Any]] | None, str | None]:
"""Return only selected JSON fields downstream; never persist CLI error output."""
try:
process = subprocess.run(["docker", *command], check=True, capture_output=True, encoding="utf-8", errors="replace", timeout=timeout)
except subprocess.TimeoutExpired:
return None, f"docker_{command[0]}_timeout"
except subprocess.CalledProcessError:
return None, f"docker_{command[0]}_failed"
except OSError:
return None, f"docker_{command[0]}_unavailable"
try:
rows = [json.loads(line) for line in process.stdout.splitlines() if line.strip()]
except (ValueError, TypeError):
return None, f"invalid_{command[0]}_json"
if any(not isinstance(row, dict) for row in rows):
return None, f"invalid_{command[0]}_json"
return rows, None
def collect_docker_sample(prefix: str, *, command_timeout: float = 5.0) -> dict[str, Any]:
"""Inventory exact name-prefix matches, then request stats for their IDs only."""
_validate_prefix(prefix)
if not math.isfinite(command_timeout) or command_timeout <= 0:
raise ValueError("command_timeout must be finite and positive")
started = time.monotonic()
sample: dict[str, Any] = {
"record_type": "resource_sample",
"timestamp_utc": datetime.now(UTC).isoformat(),
"monotonic_s": started,
"container_prefix": prefix,
"matched_containers": None,
"containers": [],
"total_memory_bytes": None,
"total_cpu_percent": None,
"errors": [],
}
def finish() -> dict[str, Any]:
sample["collection_seconds"] = time.monotonic() - started
return sample
inventory, error = _docker(["ps", "--no-trunc", "--format", "{{json .}}"], command_timeout)
if inventory is None:
sample["errors"].append(error)
return finish()
selected: dict[str, str] = {}
for row in inventory:
name = row.get("Names")
if not isinstance(name, str):
sample["errors"].append("invalid_inventory")
return finish()
if not name.startswith(prefix):
continue
container_id = row.get("ID")
if not isinstance(container_id, str) or not re.fullmatch(r"[a-f0-9]{64}", container_id) or container_id in selected:
sample["errors"].append("invalid_inventory")
return finish()
selected[container_id] = name
sample["matched_containers"] = len(selected)
if not selected:
sample["total_memory_bytes"] = 0
sample["total_cpu_percent"] = 0.0
return finish()
# An empty argument list means "all containers" to docker stats. Never call it
# until exact local selection has produced at least one validated ID.
stats, error = _docker(["stats", "--no-stream", "--format", "{{json .}}", "--", *selected], command_timeout)
if stats is None:
sample["errors"].append(error)
stats = []
for container_id, name in selected.items():
# Docker stats normally abbreviates ID even though inventory is no-trunc.
matching = [row for row in stats if isinstance(row.get("ID"), str) and re.fullmatch(r"[a-f0-9]{12,64}", row["ID"]) and container_id.startswith(row["ID"])]
stat = matching[0] if len(matching) == 1 else {}
usage = str(stat.get("MemUsage", "")).split("/")
memory = parse_bytes(usage[0]) if len(usage) == 2 else None
limit = parse_bytes(usage[1]) if len(usage) == 2 else None
cpu = _cpu_percent(str(stat.get("CPUPerc", "")))
if memory is None or limit is None or cpu is None:
sample["errors"].append("missing_or_invalid_container_stats")
sample["containers"].append({"container_id": container_id, "name": name, "memory_bytes": memory, "memory_limit_bytes": limit, "cpu_percent": cpu})
containers = sample["containers"]
if all(item["memory_bytes"] is not None for item in containers):
sample["total_memory_bytes"] = sum(item["memory_bytes"] for item in containers)
if all(item["cpu_percent"] is not None for item in containers):
sample["total_cpu_percent"] = sum(item["cpu_percent"] for item in containers)
sample["errors"] = list(dict.fromkeys(sample["errors"]))
return finish()
class DockerResourceSampler:
"""Background sampler with drain-on-exit and an exclusively created JSONL file."""
def __init__(self, prefix: str, output: str | Path, *, interval: float = 1.0, command_timeout: float = 5.0):
_validate_prefix(prefix)
if not all(math.isfinite(value) and value > 0 for value in (interval, command_timeout)):
raise ValueError("interval and command_timeout must be finite and positive")
self.prefix = prefix
self.output = Path(output)
self.interval = interval
self.command_timeout = command_timeout
self._stop = threading.Event()
self._first_sample = threading.Event()
self._thread: threading.Thread | None = None
self._file: TextIO | None = None
self._error: BaseException | None = None
self.summary: dict[str, Any] = {
"record_type": "resource_summary",
"container_prefix": prefix,
"sample_count": 0,
"incomplete_samples": 0,
"peak_sampled_total_memory_bytes": None,
"peak_sampled_total_cpu_percent": None,
"max_sampled_containers": None,
"interval_seconds": interval,
"peak_semantics": "maximum observed sample; not a guaranteed peak or host/VM total",
}
@property
def running(self) -> bool:
return self._thread is not None and self._thread.is_alive()
def _write(self, value: dict[str, Any]) -> None:
assert self._file is not None
self._file.write(json.dumps(value, allow_nan=False) + "\n")
self._file.flush()
def _run(self) -> None:
try:
while not self._stop.is_set():
sample = collect_docker_sample(self.prefix, command_timeout=self.command_timeout)
self._write(sample)
self.summary["sample_count"] += 1
self.summary["incomplete_samples"] += bool(sample["errors"])
for field, target in (("total_memory_bytes", "peak_sampled_total_memory_bytes"), ("total_cpu_percent", "peak_sampled_total_cpu_percent"), ("matched_containers", "max_sampled_containers")):
value = sample[field]
previous = self.summary[target]
if value is not None and (previous is None or value > previous):
self.summary[target] = value
self._first_sample.set()
self._stop.wait(self.interval)
except BaseException as exc:
self._error = exc
finally:
self._first_sample.set()
def start(self) -> DockerResourceSampler:
if self._thread is not None:
raise RuntimeError("A sampler instance can only be started once")
self._file = self.output.open("x", encoding="utf-8")
self._thread = threading.Thread(target=self._run, name="sandbox-resource-sampler")
try:
self._thread.start()
except BaseException:
self._file.close()
self._file = None
raise
return self
def wait_for_first_sample(self, timeout: float | None = None) -> bool:
"""Optional synchronization before a workload; never starts Docker resources."""
return self._first_sample.wait(timeout)
def stop(self) -> dict[str, Any]:
if self._thread is None or self._file is None:
return self.summary
self._stop.set()
# Each subprocess has a timeout. Drain the in-flight sample before closing
# its output; a stop must not leave the writer or a Docker client running.
self._thread.join()
try:
self.summary["timestamp_utc"] = datetime.now(UTC).isoformat()
self.summary["monotonic_s"] = time.monotonic()
self.summary["sampler_failed"] = self._error is not None
self._write(self.summary)
finally:
self._file.close()
self._file = None
if self._error is not None:
raise RuntimeError("Resource sampler failed; measurements are incomplete") from self._error
return self.summary
def __enter__(self) -> DockerResourceSampler:
return self.start()
def __exit__(self, exc_type, exc_value, traceback) -> None:
self.stop()
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("--prefix", required=True, help="Unique benchmark container prefix, ending in '-'")
parser.add_argument("--output", required=True, type=Path, help="New JSONL file; existing files are never overwritten")
parser.add_argument("--interval", type=float, default=1.0, help="Seconds between completed samples")
parser.add_argument("--command-timeout", type=float, default=5.0)
parser.add_argument("--duration", type=float, required=True, help="Bounded measurement duration in seconds")
args = parser.parse_args(argv)
if not math.isfinite(args.duration) or args.duration <= 0:
parser.error("duration must be finite and positive")
try:
sampler = DockerResourceSampler(args.prefix, args.output, interval=args.interval, command_timeout=args.command_timeout)
except ValueError as exc:
parser.error(str(exc))
try:
with sampler:
threading.Event().wait(args.duration)
except KeyboardInterrupt:
return 130
return 0
if __name__ == "__main__":
raise SystemExit(main())

View File

@ -1,5 +1,7 @@
#!/usr/bin/env python3
"""Aggregate JSONL benchmark results into summary tables.
"""Aggregate bench_provider JSONL results into summary tables.
Prewarm cases and resource samples use separate formats and are not summarized.
Usage::
@ -12,23 +14,17 @@ from __future__ import annotations
import argparse
import json
import statistics
import sys
from collections import defaultdict
from pathlib import Path
from typing import Any
def _p(arr: list[float], pct: float) -> float:
if not arr:
return 0.0
s = sorted(arr)
if len(s) == 1:
return s[0]
rank = (len(s) - 1) * (pct / 100)
lower = int(rank)
upper = min(lower + 1, len(s) - 1)
weight = rank - lower
return s[lower] * (1 - weight) + s[upper] * weight
def _p(arr: list[float], pct: int) -> float:
if len(arr) < 2:
return arr[0] if arr else 0.0
return statistics.quantiles(arr, n=100, method="inclusive")[pct - 1]
def _load_jsonl(paths: list[Path]) -> list[dict[str, Any]]:
@ -59,7 +55,7 @@ def _summarize(rows: list[dict[str, Any]], group_by: list[str]) -> list[dict[str
groups[_group_key(r, group_by)].append(r)
summary: list[dict[str, Any]] = []
for key, group in sorted(groups.items()):
for key, group in sorted(groups.items(), key=lambda item: repr(item[0])):
ok = [r for r in group if r.get("success")]
a = [r["acquire_ms"] for r in ok]
ru = [r.get("run_ms", 0) for r in ok]
@ -97,6 +93,12 @@ _COLUMNS = [
"scenario",
"workload",
"concurrency",
"users",
"no_warmpool",
"replicas",
"idle_timeout",
"health_check_skip_seconds",
"image",
"count",
"ok",
"errors",
@ -116,10 +118,11 @@ _COLUMNS = [
def _print_table(rows: list[dict[str, Any]], fmt: str = "plain") -> None:
columns = list(dict.fromkeys(_COLUMNS + [key for row in rows for key in row if key not in _COLUMNS]))
if fmt == "csv":
import csv as _csv
w = _csv.DictWriter(sys.stdout, fieldnames=_COLUMNS, extrasaction="ignore")
w = _csv.DictWriter(sys.stdout, fieldnames=columns)
w.writeheader()
w.writerows(rows)
return
@ -129,7 +132,7 @@ def _print_table(rows: list[dict[str, Any]], fmt: str = "plain") -> None:
print("(no data)")
return
headers = [c for c in _COLUMNS if any(r.get(c) is not None for r in rows)]
headers = [column for column in columns if any(row.get(column) is not None for row in rows)]
col_widths = {h: len(h) for h in headers}
for r in rows:
for h in headers:
@ -147,12 +150,12 @@ def _print_table(rows: list[dict[str, Any]], fmt: str = "plain") -> None:
def main(argv: list[str] | None = None) -> int:
p = argparse.ArgumentParser(description="Aggregate JSONL benchmark results")
p = argparse.ArgumentParser(description="Aggregate bench_provider.py JSONL results (not prewarm cases or resource samples)")
p.add_argument("inputs", nargs="+", help="JSONL file(s) from sandbox/bench_provider.py")
p.add_argument(
"--group",
default="provider,scenario,workload,concurrency",
help="Comma-separated grouping dimensions (default: provider,scenario,workload,concurrency)",
default="provider,scenario,workload,concurrency,users,no_warmpool,replicas,idle_timeout,health_check_skip_seconds,image",
help="Comma-separated grouping dimensions (default includes provider configuration and user count)",
)
p.add_argument(
"--csv",
@ -176,6 +179,19 @@ def main(argv: list[str] | None = None) -> int:
print("No valid JSONL rows found.", file=sys.stderr)
return 1
provider_rows = [row for row in rows if isinstance(row, dict) and row.get("record_type") is None and "success" in row]
if len(provider_rows) != len(rows):
print(f"Ignored {len(rows) - len(provider_rows)} non-provider record(s); summarize.py reports provider results only.", file=sys.stderr)
rows = provider_rows
if not rows:
print("No bench_provider.py result rows found. Prewarm and resource JSONL use separate formats.", file=sys.stderr)
return 1
for index, row in enumerate(rows, start=1):
missing = [field for field in ("acquire_ms", "total_ms") if field not in row]
if row.get("success") and missing:
print(f"Provider result {index} is missing timing fields: {', '.join(missing)}.", file=sys.stderr)
return 1
group_by = [g.strip() for g in args.group.split(",") if g.strip()]
summary = _summarize(rows, group_by)
@ -186,7 +202,7 @@ def main(argv: list[str] | None = None) -> int:
else:
_print_table(summary, fmt="plain")
return 0
return int(any(not row.get("success") for row in rows))
if __name__ == "__main__":

View File

@ -0,0 +1,345 @@
"""Offline real-graph checks; Events control model/startup overlap, not sleeps."""
from __future__ import annotations
import asyncio
import json
import threading
from types import SimpleNamespace
import pytest
from deerflow.config.app_config import AppConfig
from deerflow.config.paths import Paths
from deerflow.config.sandbox_config import SandboxConfig
from deerflow.sandbox.sandbox_provider import SandboxProvider, reset_sandbox_provider, set_sandbox_provider
from scripts.benchmark.sandbox import bench_prewarm as bench
class EventProvider(SandboxProvider):
def __init__(self):
self.started = asyncio.Event()
self.finish = asyncio.Event()
self.active = {}
self.released = []
self.executed = []
self._lock = threading.Lock()
self._warm_pool = {}
def acquire(self, thread_id=None, *, user_id=None):
raise AssertionError("async tool path must not fall back to sync acquire")
async def acquire_async(self, thread_id=None, *, user_id=None):
self.started.set()
# The real provider drains non-cancellable startup on cancellation.
task = asyncio.create_task(self.finish.wait())
try:
await asyncio.shield(task)
except asyncio.CancelledError:
await task
raise
sid = f"{user_id}:{thread_id}"
self.active[sid] = FakeSandbox(sid, self.executed)
return sid
def get(self, sid):
return self.active.get(sid)
def get_scoped(self, sid, *, thread_id, user_id):
return self.get(sid) if sid == f"{user_id}:{thread_id}" else None
def release(self, sid):
self.released.append(sid)
resource = self.active.pop(sid, None)
if resource:
self._warm_pool[sid] = (resource, 0)
def _active_count_locked(self):
return len(self.active)
def destroy(self, sid):
self._warm_pool.pop(sid, None)
return True
class FakeSandbox:
def __init__(self, sid, executed):
self.id = sid
self.executed = executed
def execute_command(self, command, **kwargs):
self.executed.append(command)
return bench.COMMAND_MARKER
def release_command_scope(self, owner_id):
pass
@pytest.fixture
def setup(monkeypatch, tmp_path):
config = AppConfig(sandbox=SandboxConfig(use="deerflow.community.aio_sandbox:AioSandboxProvider"))
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
monkeypatch.setattr("deerflow.config.paths._paths", Paths(str(tmp_path)))
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
monkeypatch.setattr("deerflow.config.app_config.get_app_config", lambda: config)
monkeypatch.setattr("deerflow.sandbox.tools.get_app_config", lambda: config)
provider = EventProvider()
set_sandbox_provider(provider)
yield provider, config
reset_sandbox_provider()
@pytest.fixture
def observed(setup):
provider, _ = setup
with bench.observe_provider(provider):
yield setup
async def run_after_prewarm_ready(provider, config, *, case, thread_id):
"""Release the scripted first response only after the speculative lease exists."""
trace, gate = bench.Trace(), asyncio.Event()
task = asyncio.create_task(bench.run_case(provider, config, case=case, prewarm=True, model_delay=0, thread_id=thread_id, user_id="alice", trace=trace, model_gate=gate))
try:
await asyncio.wait_for(trace.model_started.wait(), 2)
await asyncio.wait_for(trace.acquire_ready.wait(), 2)
gate.set()
return await asyncio.wait_for(task, 3)
finally:
provider.finish.set()
gate.set()
await asyncio.gather(task, return_exceptions=True)
@pytest.mark.anyio
async def test_real_graph_prewarm_starts_during_model_and_is_claimed_once(observed):
provider, config = observed
model_gate = asyncio.Event()
trace = bench.Trace()
task = asyncio.create_task(bench.run_case(provider, config, case="tool", prewarm=True, model_delay=0, thread_id="overlap", user_id="alice", trace=trace, model_gate=model_gate))
try:
await asyncio.wait_for(trace.model_started.wait(), 2)
await asyncio.wait_for(provider.started.wait(), 2)
assert not provider.executed
assert not trace.tool_requested.is_set()
provider.finish.set()
await asyncio.wait_for(trace.acquire_ready.wait(), 2)
model_gate.set()
row = await asyncio.wait_for(task, 3)
finally:
provider.finish.set()
model_gate.set()
await asyncio.gather(task, return_exceptions=True)
assert row["success"], json.dumps(row)
assert row["acquire_calls"] == 1
assert row["prewarm_reused"] is True
assert row["tool_output_valid"] is True
assert provider.executed and provider.released == ["alice:overlap"]
assert not provider.active
@pytest.mark.anyio
async def test_real_graph_baseline_waits_until_tool_to_start(observed):
provider, config = observed
model_gate = asyncio.Event()
trace = bench.Trace()
task = asyncio.create_task(bench.run_case(provider, config, case="tool", prewarm=False, model_delay=0, thread_id="lazy", user_id="alice", trace=trace, model_gate=model_gate))
try:
await asyncio.wait_for(trace.model_started.wait(), 2)
assert not provider.started.is_set()
model_gate.set()
await asyncio.wait_for(provider.started.wait(), 2)
assert trace.tool_requested.is_set()
provider.finish.set()
row = await asyncio.wait_for(task, 3)
finally:
model_gate.set()
provider.finish.set()
await asyncio.gather(task, return_exceptions=True)
assert row["success"] and row["acquire_calls"] == 1, json.dumps(row)
assert row["prewarm_reused"] is False
@pytest.mark.anyio
@pytest.mark.parametrize("case,expected_acquires", [("plain", 0), ("fenced_no_tool", 1), ("model_error", 1)])
async def test_no_tool_and_failure_close_actual_graph_ownership(observed, case, expected_acquires):
provider, config = observed
provider.finish.set()
if expected_acquires:
row = await run_after_prewarm_ready(provider, config, case=case, thread_id=case)
else:
row = await bench.run_case(provider, config, case=case, prewarm=True, model_delay=0, thread_id=case, user_id="alice")
assert row["success"], json.dumps(row)
assert row["acquire_calls"] == expected_acquires
assert row["unused_prewarm"] is bool(expected_acquires)
assert not provider.active and not provider.executed
assert row["lease_remaining"] is False
@pytest.mark.anyio
@pytest.mark.parametrize("prewarm", [False, True])
async def test_cancelled_graph_drains_started_acquire_before_reporting_cleanup(observed, prewarm):
provider, config = observed
trace = bench.Trace()
task = asyncio.create_task(bench.run_case(provider, config, case="cancel", prewarm=prewarm, model_delay=0, thread_id="cancel", user_id="alice", trace=trace))
try:
await asyncio.wait_for(trace.cancel_requested.wait(), 2)
assert not task.done()
provider.finish.set()
row = await asyncio.wait_for(task, 3)
finally:
provider.finish.set()
await asyncio.gather(task, return_exceptions=True)
assert row["success"] and row["outcome"] == "cancelled", json.dumps(row)
assert row["lease_remaining"] is False
assert not provider.active
@pytest.mark.anyio
async def test_tool_semantic_failure_is_not_a_successful_latency_row(observed, monkeypatch):
provider, config = observed
provider.finish.set()
monkeypatch.setattr(FakeSandbox, "execute_command", lambda self, command, **kwargs: "Error: injected command failure")
row = await bench.run_case(provider, config, case="tool", prewarm=False, model_delay=0, thread_id="bad-tool", user_id="alice")
assert row["success"] is False
assert row["tool_output_valid"] is False
assert row["lease_remaining"] is False
@pytest.mark.anyio
async def test_two_users_keep_trace_and_execution_ownership_separate(observed):
provider, config = observed
provider.finish.set()
rows = await asyncio.gather(*[bench.run_case(provider, config, case="tool", prewarm=True, model_delay=0, thread_id="same-thread", user_id=user) for user in ("alice", "bob")])
assert all(row["success"] for row in rows), rows
assert [row["sandbox_ids"] for row in rows] == [["alice:same-thread"], ["bob:same-thread"]]
assert all(row["acquire_calls"] == 1 and row["release_calls"] == 1 for row in rows)
assert not provider.active
@pytest.mark.anyio
async def test_matrix_retains_each_variant_and_cleans_exact_batch_resources(setup, tmp_path):
provider, config = setup
provider.finish.set()
args = SimpleNamespace(seed=5002, cases=["tool"], model_delays=[0], iterations=1, concurrency=2, users=2, run_id="matrix")
path = tmp_path / "rows.jsonl"
with path.open("w", encoding="utf-8") as stream:
assert await bench.run_matrix(provider, config, args, stream)
records = [json.loads(line) for line in path.read_text().splitlines()]
rows = [row for row in records if row["record_type"] == "prewarm_case"]
assert len(rows) == 4
assert all(row["acquire_calls"] == 1 and row["release_calls"] == 1 for row in rows)
assert {row["prewarm"] for row in rows} == {False, True}
assert len({row["thread_id"] for row in rows}) == 4
assert all(row["started_monotonic_s"] < row["finished_monotonic_s"] for row in rows)
assert not provider.active and not provider._warm_pool
resets = [row for row in records if row["record_type"] == "batch_cleanup"]
assert len(resets) == 2 and all(row["success"] for row in resets)
@pytest.mark.parametrize("interval", ["0", "nan", "-1"])
def test_invalid_resource_interval_rejected_before_provider_creation(monkeypatch, tmp_path, interval):
monkeypatch.setattr(bench, "_make_aio_provider", lambda config: pytest.fail("invalid arguments must not provision containers"))
with pytest.raises(SystemExit) as exc:
bench.main(["--output", str(tmp_path / "rows.jsonl"), "--resource-interval", interval])
assert exc.value.code == 2
@pytest.mark.anyio
async def test_expected_model_failure_cannot_mask_cleanup_failure(observed, monkeypatch):
provider, config = observed
provider.finish.set()
def fail_release(sid):
raise OSError("injected cleanup failure")
monkeypatch.setattr(provider, "release", fail_release)
row = await run_after_prewarm_ready(provider, config, case="model_error", thread_id="cleanup-error")
assert row["success"] is False
assert row["cleanup_error_type"] == "OSError"
@pytest.mark.anyio
async def test_failed_acquire_records_its_own_finish_time(setup, monkeypatch):
provider, _ = setup
async def fail(*args, **kwargs):
raise RuntimeError("injected acquire failure")
monkeypatch.setattr(provider, "acquire_async", fail)
trace = bench.Trace()
token = bench._TRACE.set(trace)
try:
with bench.observe_provider(provider):
with pytest.raises(RuntimeError, match="injected acquire failure"):
await provider.acquire_async("thread", user_id="alice")
entry = trace.acquisitions[0]
assert entry["finished"] >= entry["started"]
assert "ready" not in entry
finally:
bench._TRACE.reset(token)
@pytest.mark.anyio
async def test_failed_batch_destroy_is_recorded_and_stops_matrix(setup, monkeypatch, tmp_path):
provider, config = setup
provider.finish.set()
def fail_destroy(*args):
raise OSError("injected destroy failure")
monkeypatch.setattr(bench, "_evict_from_warm", fail_destroy)
args = SimpleNamespace(seed=5002, cases=["tool"], model_delays=[0], iterations=1, concurrency=1, users=1, run_id="bad-reset")
path = tmp_path / "rows.jsonl"
with path.open("w", encoding="utf-8") as stream:
assert not await bench.run_matrix(provider, config, args, stream)
rows = [json.loads(line) for line in path.read_text().splitlines()]
assert len(rows) == 2
assert rows[0]["record_type"] == "prewarm_case" and rows[0]["success"]
assert rows[1]["record_type"] == "batch_cleanup" and not rows[1]["success"]
assert rows[1]["errors"] == [{"sandbox_id": rows[0]["sandbox_ids"][0], "error_type": "OSError"}]
def test_non_aio_config_rejected_before_provisioning(monkeypatch, tmp_path):
config = AppConfig(sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider"))
monkeypatch.setattr(bench, "get_app_config", lambda: config)
monkeypatch.setattr(bench, "_make_aio_provider", lambda config: pytest.fail("must not provision for a mismatched treatment"))
with pytest.raises(SystemExit) as exc:
bench.main(["--output", str(tmp_path / "rows.jsonl")])
assert exc.value.code == 2
def test_shutdown_failure_still_stops_sampler_and_propagates(monkeypatch, tmp_path):
events = []
config = AppConfig(sandbox=SandboxConfig(use="deerflow.community.aio_sandbox:AioSandboxProvider"))
monkeypatch.setattr(bench, "get_app_config", lambda: config)
monkeypatch.setattr(bench, "_make_aio_provider", lambda config: (object(), {"image": "offline"}))
monkeypatch.setattr(bench, "set_sandbox_provider", lambda provider: None)
monkeypatch.setattr(bench.subprocess, "run", lambda *args, **kwargs: SimpleNamespace(stdout="revision" if kwargs.get("text") else b""))
class Sampler:
def __init__(self, *args, **kwargs):
pass
def start(self):
events.append("start")
def wait_for_first_sample(self, **kwargs):
return True
def stop(self):
events.append("stop")
async def run(*args):
return True
def failed_shutdown():
events.append("shutdown")
raise OSError("injected shutdown failure")
monkeypatch.setattr(bench, "DockerResourceSampler", Sampler)
monkeypatch.setattr(bench, "run_matrix", run)
monkeypatch.setattr(bench, "shutdown_sandbox_provider", failed_shutdown)
with pytest.raises(OSError, match="injected shutdown failure"):
bench.main(["--output", str(tmp_path / "results.jsonl")])
assert events == ["start", "shutdown", "stop"]

View File

@ -1,6 +1,7 @@
from __future__ import annotations
import importlib.util
import json
import os
import sys
from pathlib import Path
@ -325,3 +326,334 @@ def test_boxlite_shim_workaround_loud_fails_for_other_versions(monkeypatch, tmp_
"sandbox-id",
boxes_dir=str(boxes_dir),
)
class _FakeAioProvider(_FakeProvider):
"""AIO warm entries are metadata, not the closable execution client."""
def __init__(self):
super().__init__()
self._sandboxes = {}
self.destroyed = []
self.crashed = []
self.alive = set()
self._backend = bench.types.SimpleNamespace(destroy=self._crash, is_alive=lambda info: info.sandbox_id in self.alive)
def _crash(self, info):
self.crashed.append(info.sandbox_id)
self.alive.discard(info.sandbox_id)
def acquire(self, thread_id=None, *, user_id=None):
sid = f"actual-24-char-id:{user_id}:{thread_id}"
with self._lock:
self._warm_pool.pop(sid, None)
self._sandboxes[sid] = self._sandbox
self.alive.add(sid)
return sid
def release(self, sid):
with self._lock:
self._sandboxes.pop(sid)
self._warm_pool[sid] = (bench.types.SimpleNamespace(sandbox_id=sid), 0)
def destroy(self, sid):
with self._lock:
self.destroyed.append(sid)
self._warm_pool.pop(sid, None)
self.alive.discard(sid)
return True
def _active_count_locked(self):
return len(self._sandboxes)
def test_aio_no_warmpool_destroys_container_not_metadata():
provider = _FakeAioProvider()
result = bench._run_one_turn(provider, "aio-docker", "cold_unique_thread", "noop", "true", 0, 1, "user", "thread", True)
assert result.success
assert provider.destroyed == ["actual-24-char-id:user:thread"]
assert provider.alive == set()
assert provider._warm_pool == {}
def test_deferred_destroy_is_failed_measurement():
provider = _FakeAioProvider()
provider.destroy = lambda sid: False
result = bench._run_one_turn(provider, "aio-docker", "cold_unique_thread", "noop", "true", 0, 1, "user", "thread", True)
assert not result.success
assert "destroy" in result.error.lower()
assert provider.alive
assert provider._warm_pool # retain metadata so final shutdown can still clean it
@pytest.mark.parametrize("command_raises", [False, True])
@pytest.mark.parametrize("destroy_raises", [False, True])
def test_failed_turn_separates_execution_release_and_cleanup_times(monkeypatch, command_raises, destroy_raises):
clock = [0.0]
monkeypatch.setattr(bench.time, "perf_counter", lambda: clock[0])
provider = _FakeAioProvider()
original_acquire = provider.acquire
original_release = provider.release
original_destroy = provider.destroy
def acquire(*args, **kwargs):
clock[0] += 0.010
return original_acquire(*args, **kwargs)
def execute(command, timeout=None):
clock[0] += 0.020
if command_raises:
raise RuntimeError("command failed")
return "Error: command failed"
def release(sid):
clock[0] += 0.030
return original_release(sid)
def destroy(sid):
clock[0] += 0.040
if destroy_raises:
raise RuntimeError("destroy failed")
return original_destroy(sid)
monkeypatch.setattr(provider, "acquire", acquire)
monkeypatch.setattr(provider._sandbox, "execute_command", execute)
monkeypatch.setattr(provider, "release", release)
monkeypatch.setattr(provider, "destroy", destroy)
result = bench._run_one_turn(provider, "aio-docker", "cold_unique_thread", "noop", "true", 0, 1, "user", "thread", True)
assert not result.success
assert "command failed" in result.error
assert ("destroy failed" in result.error) is destroy_raises
assert result.acquire_ms == pytest.approx(10)
assert result.run_ms == pytest.approx(20)
assert result.release_ms == pytest.approx(30)
assert result.cleanup_ms == pytest.approx(40)
assert result.total_ms == pytest.approx(100)
def test_failed_acquire_records_elapsed_time_without_fabricating_later_stages(monkeypatch):
clock = [0.0]
monkeypatch.setattr(bench.time, "perf_counter", lambda: clock[0])
provider = _FakeAioProvider()
def acquire(*args, **kwargs):
clock[0] += 0.125
raise RuntimeError("acquire failed")
monkeypatch.setattr(provider, "acquire", acquire)
result = bench._run_one_turn(provider, "aio-docker", "cold_unique_thread", "noop", "true", 0, 1, "user", "thread", True)
assert not result.success
assert result.acquire_ms == pytest.approx(125)
assert result.total_ms == pytest.approx(125)
assert result.run_ms == result.release_ms == result.cleanup_ms == 0
assert provider.destroyed == []
def test_fault_injection_uses_returned_aio_identity(tmp_path):
provider = _FakeAioProvider()
rows = bench._run_scenario(provider, "aio-docker", "warm_same_thread", "noop", 1, 1, tmp_path / "out.jsonl", False, {}, fault_inject_after=0)
assert rows[0].success
assert provider.crashed == [rows[0].sandbox_id]
assert rows[0].sandbox_id in provider._warm_pool # recovery must discover the dead entry
assert not provider.alive
def test_replica_pressure_holds_concurrent_aio_clients_and_records_users(monkeypatch, tmp_path):
provider = _FakeAioProvider()
monkeypatch.setitem(bench.PROVIDER_FACTORIES, "aio-docker", lambda config: (provider, config))
output = tmp_path / "out.jsonl"
rc = bench.main(["--provider", "aio-docker", "--scenario", "replica_pressure", "--iterations", "4", "--concurrency", "2", "--users", "2", "--replicas", "1", "--warmup-iterations", "0", "--output", str(output)])
rows = [json.loads(line) for line in output.read_text().splitlines()]
assert rc == 0
assert len(rows) == 4
assert len({r["user_id"] for r in rows}) == 2
assert all(r["active_at_acquire"] == 2 for r in rows)
assert all(r["concurrency"] == 2 for r in rows)
assert provider.shutdown_called
def test_failed_workload_cli_exits_nonzero_and_keeps_rows(monkeypatch, tmp_path):
provider = _FakeProvider(_FakeSandbox("Error: command failed"))
monkeypatch.setitem(bench.PROVIDER_FACTORIES, "aio-docker", lambda config: (provider, config))
output = tmp_path / "out.jsonl"
rc = bench.main(["--provider", "aio-docker", "--iterations", "1", "--warmup-iterations", "0", "--output", str(output)])
assert rc == 1
assert json.loads(output.read_text())["success"] is False
assert provider.shutdown_called
def test_failed_warmup_stops_benchmark(monkeypatch, tmp_path):
provider = _FakeProvider(_FakeSandbox("Error: not ready"))
monkeypatch.setitem(bench.PROVIDER_FACTORIES, "aio-docker", lambda config: (provider, config))
rc = bench.main(["--provider", "aio-docker", "--iterations", "1", "--output", str(tmp_path / "out.jsonl")])
assert rc == 1
assert provider.shutdown_called
def test_default_summary_does_not_merge_cold_baseline_with_warm_pool(tmp_path, capsys):
row = {"provider": "aio-docker", "scenario": "warm_same_thread", "workload": "noop", "concurrency": 1, "success": True, "acquire_ms": 1, "total_ms": 2}
path = tmp_path / "results.jsonl"
path.write_text("\n".join(json.dumps({**row, "no_warmpool": disabled}) for disabled in (True, False)))
assert summarize.main([str(path), "--json"]) == 0
rows = json.loads(capsys.readouterr().out)
assert len(rows) == 2
assert {r["no_warmpool"] for r in rows} == {True, False}
def test_summary_failed_data_exits_nonzero(tmp_path, capsys):
path = tmp_path / "results.jsonl"
path.write_text(json.dumps({"provider": "aio-docker", "success": False}))
assert summarize.main([str(path), "--json"]) == 1
assert json.loads(capsys.readouterr().out)[0]["errors"] == 1
@pytest.mark.parametrize("success", [True, False])
def test_summary_ignores_non_provider_records(tmp_path, capsys, success):
records = [
{"record_type": "configuration", "prewarm": True},
{"record_type": "prewarm_case", "success": True},
{"record_type": "batch_cleanup", "success": False},
{"provider": "aio-docker", "success": success, "acquire_ms": 1, "total_ms": 2},
]
path = tmp_path / "mixed.jsonl"
path.write_text("\n".join(json.dumps(row) for row in records), encoding="utf-8")
assert summarize.main([str(path), "--json"]) == int(not success)
output = capsys.readouterr()
summary = json.loads(output.out)
assert len(summary) == 1
assert summary[0]["count"] == 1
assert summary[0]["errors"] == int(not success)
assert "non-provider" in output.err
@pytest.mark.parametrize(
"records",
[
[{"record_type": "configuration"}],
[{"record_type": "prewarm_case", "success": True}],
[{"record_type": "configuration"}, {"record_type": "prewarm_case", "success": True}, {"record_type": "batch_cleanup", "success": True}],
],
ids=["configuration-only", "prewarm-case-only", "prewarm-run"],
)
def test_summary_rejects_prewarm_only_input_without_traceback(tmp_path, capsys, records):
path = tmp_path / "prewarm.jsonl"
path.write_text("\n".join(json.dumps(row) for row in records), encoding="utf-8")
assert summarize.main([str(path), "--json"]) == 1
output = capsys.readouterr()
assert output.out == ""
assert "bench_provider.py" in output.err
@pytest.mark.parametrize("missing", ["acquire_ms", "total_ms"])
def test_summary_reports_incomplete_provider_timings(tmp_path, capsys, missing):
row = {"provider": "aio-docker", "success": True, "acquire_ms": 1, "total_ms": 2}
row.pop(missing)
path = tmp_path / "incomplete.jsonl"
path.write_text(json.dumps(row), encoding="utf-8")
assert summarize.main([str(path), "--json"]) == 1
output = capsys.readouterr()
assert output.out == ""
assert missing in output.err
@pytest.mark.parametrize(
"output",
[
"traceback\nExit Code: 1",
"Command exited with code 1",
"Command was terminated because its shell session ended.",
"Command produced no output change for 30 seconds; it may still be running. Command outcome is unknown and was not retried.",
],
)
def test_provider_failure_text_cannot_be_a_successful_measurement(output):
provider = _FakeProvider(_FakeSandbox(output))
result = bench._run_one_turn(provider, "aio-docker", "warm_same_thread", "noop", "true", 0, 1, "user", "thread", False)
assert not result.success
assert provider.released == ["sandbox-id"]
def test_failed_pressure_acquire_releases_peers_without_barrier_timeout(tmp_path):
provider = _FakeAioProvider()
acquire = provider.acquire
def failing_acquire(thread_id, *, user_id):
if thread_id.endswith("pressure-1"):
raise RuntimeError("injected create failure")
return acquire(thread_id, user_id=user_id)
provider.acquire = failing_acquire
rows = bench._run_scenario(provider, "aio-docker", "replica_pressure", "noop", 2, 2, tmp_path / "out.jsonl", False, {"replicas": 1})
assert all(not row.success for row in rows)
assert not provider._sandboxes
def test_partial_pressure_batch_preserves_actual_active_count(tmp_path):
rows = bench._run_scenario(_FakeAioProvider(), "aio-docker", "replica_pressure", "noop", 3, 2, tmp_path / "out.jsonl", False, {"replicas": 1})
assert all(row.success for row in rows)
assert [row.active_at_acquire for row in rows] == [2, 2, 1]
def test_new_invocation_cannot_reuse_previous_thread_identity(tmp_path):
provider = _FakeAioProvider()
args = (provider, "aio-docker", "cold_unique_thread", "noop", 1, 1, tmp_path / "out.jsonl", False, {})
first = bench._run_scenario(*args)[0]
second = bench._run_scenario(*args)[0]
assert first.thread_id != second.thread_id
assert first.sandbox_id != second.sandbox_id
def test_aio_factory_uses_current_config_contract_and_reports_effective_image(monkeypatch):
import deerflow.community.aio_sandbox.aio_sandbox_provider as provider_module
real_provider = provider_module.AioSandboxProvider
original_get_config = provider_module.get_app_config
class ConfigOnlyProvider:
_positive_float = staticmethod(real_provider._positive_float)
def __init__(self):
self._config = real_provider._load_config(self)
monkeypatch.setattr(provider_module, "AioSandboxProvider", ConfigOnlyProvider)
provider, config = bench._make_aio_provider({"container_prefix": "deerflow-bench-test"})
assert provider._config["network"]["mode"] == "open"
assert config["image"] == provider._config["image"]
assert config["image"]
assert provider_module.get_app_config is original_get_config
def test_console_percentiles_match_json_summary_for_two_samples(capsys):
results = [
bench.BenchResult(
provider="aio-docker",
scenario="warm_same_thread",
workload="noop",
iteration=i,
concurrency=1,
thread_id="thread",
user_id="user",
acquire_ms=acquire_ms,
run_ms=4,
release_ms=1,
total_ms=acquire_ms + 5,
)
for i, acquire_ms in enumerate((3435, 25))
]
bench._print_summary(results, bench.types.SimpleNamespace(output="results.jsonl"))
console = capsys.readouterr().err
report = summarize._summarize([bench.asdict(result) for result in results], ["provider"])[0]
assert report["acquire_p50"] == 1730.0
for metric in ("acquire", "total"):
percentiles = " ".join(f"p{pct}={report[f'{metric}_p{pct}']:.1f}ms" for pct in (50, 95, 99))
assert percentiles in console

View File

@ -0,0 +1,240 @@
from __future__ import annotations
import importlib.util
import json
import subprocess
import sys
import threading
from pathlib import Path
from types import SimpleNamespace
import pytest
def _load_module():
path = Path(__file__).resolve().parents[1] / "scripts/benchmark/sandbox/resources.py"
spec = importlib.util.spec_from_file_location("bench_sandbox_resources", path)
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
resources = _load_module()
PREFIX = "deerflow-bench-abc123-"
OWN_ID = "a" * 64
OTHER_ID = "b" * 64
def _jsonl(*rows):
return "\n".join(json.dumps(row) for row in rows)
@pytest.mark.parametrize("text,expected", [("0B", 0), ("1.5kB", 1500), ("2 MB", 2_000_000), ("2MiB", 2 * 1024**2), ("1.25GiB", int(1.25 * 1024**3)), ("1TiB", 1024**4)])
def test_docker_memory_units(text, expected):
assert resources.parse_bytes(text) == expected
@pytest.mark.parametrize("text", ["", "N/A", "nanMiB", "-1MiB", "1.0unknown", "infB"])
def test_missing_or_invalid_memory_is_unknown(text):
assert resources.parse_bytes(text) is None
@pytest.mark.parametrize("prefix", ["", "-", "deerflow", "bad prefix-", "*all*"])
def test_sampler_requires_explicit_bounded_prefix(prefix, tmp_path):
with pytest.raises(ValueError, match="prefix"):
resources.DockerResourceSampler(prefix, tmp_path / "resources.jsonl")
def test_inventory_filters_exact_prefix_and_stats_targets_only_owned_ids(monkeypatch):
calls = []
def fake_run(command, **kwargs):
calls.append((command, kwargs))
if command[1] == "ps":
return SimpleNamespace(stdout=_jsonl({"ID": OWN_ID, "Names": PREFIX + "one"}, {"ID": OTHER_ID, "Names": "production"}, {"ID": "c" * 64, "Names": "other-" + PREFIX + "two"}))
return SimpleNamespace(stdout=_jsonl({"ID": OWN_ID[:12], "Name": PREFIX + "one", "MemUsage": "1.5MiB / 8GiB", "CPUPerc": "123.45%"}))
monkeypatch.setattr(resources.subprocess, "run", fake_run)
result = resources.collect_docker_sample(PREFIX, command_timeout=2)
assert [command[1] for command, _ in calls] == ["ps", "stats"]
assert calls[1][0][-2:] == ["--", OWN_ID]
assert OTHER_ID not in calls[1][0]
assert all(kwargs == {"check": True, "capture_output": True, "encoding": "utf-8", "errors": "replace", "timeout": 2} for _, kwargs in calls)
assert result["matched_containers"] == 1
assert result["total_memory_bytes"] == 1_572_864
assert result["total_cpu_percent"] == 123.45
assert result["containers"][0]["memory_limit_bytes"] == 8 * 1024**3
assert result["errors"] == []
assert result["timestamp_utc"].endswith("+00:00")
assert result["monotonic_s"] > 0
def test_known_empty_inventory_records_zero_without_unscoped_stats(monkeypatch):
calls = []
def fake_run(command, **kwargs):
calls.append(command)
return SimpleNamespace(stdout="")
monkeypatch.setattr(resources.subprocess, "run", fake_run)
sample = resources.collect_docker_sample(PREFIX)
assert len(calls) == 1
assert sample["matched_containers"] == 0
assert sample["total_memory_bytes"] == 0
assert sample["total_cpu_percent"] == 0
assert sample["errors"] == []
@pytest.mark.parametrize(
"failure,code", [(subprocess.CalledProcessError(1, "docker", stderr="secret credential"), "docker_ps_failed"), (subprocess.TimeoutExpired("docker", 1), "docker_ps_timeout"), (FileNotFoundError("private path"), "docker_ps_unavailable")]
)
def test_inventory_failure_is_unknown_and_does_not_leak_command_output(monkeypatch, failure, code):
def fail(*args, **kwargs):
raise failure
monkeypatch.setattr(resources.subprocess, "run", fail)
sample = resources.collect_docker_sample(PREFIX)
assert sample["matched_containers"] is None
assert sample["total_memory_bytes"] is None
assert sample["total_cpu_percent"] is None
assert sample["errors"] == [code]
assert "secret" not in json.dumps(sample)
assert "private path" not in json.dumps(sample)
@pytest.mark.parametrize("stats", ["", "{broken json", _jsonl({"ID": OTHER_ID, "Name": "production", "MemUsage": "2MiB / 8GiB", "CPUPerc": "50%"}), _jsonl({"ID": OWN_ID[:12], "MemUsage": "N/A", "CPUPerc": "--"})])
def test_missing_stats_never_become_zero_or_count_other_containers(monkeypatch, stats):
def fake_run(command, **kwargs):
return SimpleNamespace(stdout=_jsonl({"ID": OWN_ID, "Names": PREFIX + "one"}) if command[1] == "ps" else stats)
monkeypatch.setattr(resources.subprocess, "run", fake_run)
sample = resources.collect_docker_sample(PREFIX)
assert sample["matched_containers"] == 1
assert sample["total_memory_bytes"] is None
assert sample["total_cpu_percent"] is None
assert sample["errors"]
assert all(container["container_id"] == OWN_ID for container in sample["containers"])
def test_sampler_drains_inflight_sample_and_records_sampled_peak(monkeypatch, tmp_path):
entered = threading.Event()
release = threading.Event()
stop_entered = threading.Event()
stopped = threading.Event()
sample_calls = []
def collect(prefix, *, command_timeout):
sample_calls.append(prefix)
entered.set()
assert release.wait(5)
return {"record_type": "resource_sample", "containers": [], "matched_containers": 2, "total_memory_bytes": 2048, "total_cpu_percent": 50.0, "errors": []}
monkeypatch.setattr(resources, "collect_docker_sample", collect)
output = tmp_path / "resources.jsonl"
sampler = resources.DockerResourceSampler(PREFIX, output, interval=60)
sampler.start()
assert entered.wait(5)
def stop():
stop_entered.set()
sampler.stop()
stopped.set()
stopper = threading.Thread(target=stop)
stopper.start()
try:
assert stop_entered.wait(5)
assert not stopped.is_set()
finally:
release.set()
stopper.join(5)
sampler.stop()
assert stopped.is_set()
assert sample_calls == [PREFIX]
rows = [json.loads(line) for line in output.read_text(encoding="utf-8").splitlines()]
assert len(rows) == 2
assert rows[0]["record_type"] == "resource_sample"
summary = rows[1]
assert summary["record_type"] == "resource_summary"
assert summary["sample_count"] == 1
assert summary["peak_sampled_total_memory_bytes"] == 2048
assert summary["peak_sampled_total_cpu_percent"] == 50
assert summary["max_sampled_containers"] == 2
assert not sampler.running
def test_summary_preserves_unknown_measurements_and_output_is_exclusive(monkeypatch, tmp_path):
monkeypatch.setattr(resources, "collect_docker_sample", lambda *args, **kwargs: {"record_type": "resource_sample", "matched_containers": None, "total_memory_bytes": None, "total_cpu_percent": None, "errors": ["docker_ps_failed"]})
output = tmp_path / "resources.jsonl"
with resources.DockerResourceSampler(PREFIX, output, interval=60) as sampler:
sampler.wait_for_first_sample(timeout=5)
summary = json.loads(output.read_text(encoding="utf-8").splitlines()[-1])
assert summary["peak_sampled_total_memory_bytes"] is None
assert summary["peak_sampled_total_cpu_percent"] is None
assert summary["max_sampled_containers"] is None
assert summary["incomplete_samples"] == 1
with pytest.raises(FileExistsError):
resources.DockerResourceSampler(PREFIX, output).start()
def test_invalid_inventory_does_not_dispatch_stats(monkeypatch):
calls = []
def fake_run(command, **kwargs):
calls.append(command)
return SimpleNamespace(stdout=_jsonl({"ID": "--all", "Names": PREFIX + "one"}))
monkeypatch.setattr(resources.subprocess, "run", fake_run)
sample = resources.collect_docker_sample(PREFIX)
assert len(calls) == 1
assert sample["matched_containers"] is None
assert sample["errors"] == ["invalid_inventory"]
def test_stats_timeout_retains_inventory_but_not_fabricated_resource_values(monkeypatch):
def fake_run(command, **kwargs):
if command[1] == "ps":
return SimpleNamespace(stdout=_jsonl({"ID": OWN_ID, "Names": PREFIX + "one"}))
raise subprocess.TimeoutExpired(command, 1)
monkeypatch.setattr(resources.subprocess, "run", fake_run)
sample = resources.collect_docker_sample(PREFIX)
assert sample["matched_containers"] == 1
assert sample["total_memory_bytes"] is None
assert sample["total_cpu_percent"] is None
assert "docker_stats_timeout" in sample["errors"]
@pytest.mark.parametrize("duration", ["0", "-1", "inf", "nan"])
def test_cli_requires_bounded_positive_duration(duration, tmp_path):
output = tmp_path / "out.jsonl"
with pytest.raises(SystemExit) as exc_info:
resources.main(["--prefix", PREFIX, "--output", str(output), "--duration", duration])
assert exc_info.value.code == 2
assert not output.exists()
def test_cli_interrupt_exits_sampler_context_before_returning(monkeypatch, tmp_path):
calls = []
class Sampler:
def __init__(self, *args, **kwargs):
pass
def __enter__(self):
calls.append("start")
return self
def __exit__(self, *args):
calls.append("stop")
class Interrupt:
def wait(self, duration):
raise KeyboardInterrupt
monkeypatch.setattr(resources, "DockerResourceSampler", Sampler)
monkeypatch.setattr(resources.threading, "Event", Interrupt)
assert resources.main(["--prefix", PREFIX, "--output", str(tmp_path / "out.jsonl"), "--duration", "60"]) == 130
assert calls == ["start", "stop"]

View File

@ -134,6 +134,11 @@ class SandboxMessages(MessagesState):
ids=["unit-with-empty-doc", "doc-with-empty-unit", "ignored", "failed-unit", "failed-doc", "filtered"],
)
async def test_native_cargo_output_reaches_parent_acceptance_through_tool_node(tmp_path, monkeypatch, executor_module, source, selection, expected):
# LocalSandbox starts an independent subprocess without inheriting make's
# jobserver FDs. Stale flags produce long Cargo/rustdoc warnings that can
# displace the test summaries from the intentionally bounded evidence tail.
for variable in ("MAKEFLAGS", "MFLAGS", "CARGO_MAKEFLAGS"):
monkeypatch.delenv(variable, raising=False)
(tmp_path / "src").mkdir()
(tmp_path / "Cargo.toml").write_text('[package]\nname = "acceptance_fixture"\nversion = "0.1.0"\nedition = "2021"\n', encoding="utf-8")
(tmp_path / "src/lib.rs").write_text(source, encoding="utf-8")

View File

@ -4494,6 +4494,8 @@ class TestInjectAuthenticatedUserContextAuthz:
section: {
"sandbox_lease_owner_id": "forged-owner",
"sandbox_command_scope_id": "forged-scope",
"__deerflow_sandbox_prewarm": "forged-prewarm",
"__deerflow_sandbox_prewarm_cleanup": True,
}
},
request,
@ -4501,6 +4503,8 @@ class TestInjectAuthenticatedUserContextAuthz:
assert "sandbox_lease_owner_id" not in config[section]
assert "sandbox_command_scope_id" not in config[section]
assert "__deerflow_sandbox_prewarm" not in config[section]
assert "__deerflow_sandbox_prewarm_cleanup" not in config[section]
def test_session_body_context_cannot_inject_channel_user_id(self):
request = _make_request_with_auth_source("session")

View File

@ -635,6 +635,7 @@ def test_install_runtime_context_removes_caller_sandbox_execution_identities():
"context": {
SANDBOX_LEASE_OWNER_CONTEXT_KEY: "forged-owner",
SANDBOX_COMMAND_SCOPE_CONTEXT_KEY: "forged-scope",
"__deerflow_sandbox_prewarm": "forged-prewarm",
}
}
@ -648,6 +649,7 @@ def test_install_runtime_context_removes_caller_sandbox_execution_identities():
assert SANDBOX_LEASE_OWNER_CONTEXT_KEY not in config["context"]
assert SANDBOX_COMMAND_SCOPE_CONTEXT_KEY not in config["context"]
assert "__deerflow_sandbox_prewarm" not in config["context"]
def test_build_runtime_context_ignores_caller_project_context_key():
@ -2808,12 +2810,14 @@ def test_build_runtime_context_ignores_caller_sandbox_execution_identities():
caller_context = {
SANDBOX_LEASE_OWNER_CONTEXT_KEY: "forged-owner",
SANDBOX_COMMAND_SCOPE_CONTEXT_KEY: "forged-scope",
"__deerflow_sandbox_prewarm": "forged-prewarm",
}
ctx = _build_runtime_context("thread-1", "run-1", caller_context)
assert SANDBOX_LEASE_OWNER_CONTEXT_KEY not in ctx
assert SANDBOX_COMMAND_SCOPE_CONTEXT_KEY not in ctx
assert "__deerflow_sandbox_prewarm" not in ctx
def test_build_runtime_context_ignores_caller_audit_attribution_and_recorders():

View File

@ -2,6 +2,7 @@ from __future__ import annotations
import asyncio
import threading
from types import SimpleNamespace
from typing import get_type_hints
import pytest
@ -16,15 +17,18 @@ from langgraph.types import Command, Overwrite
from deerflow.agents.thread_state import ThreadState
from deerflow.sandbox.exceptions import SandboxAuthorizationError, SandboxRuntimeError
from deerflow.sandbox.lease import (
ensure_sandbox_lease_owner,
get_sandbox_lease_manager,
register_sandbox_prewarm_cleanup,
release_sandbox_execution_lease,
release_sandbox_execution_lease_async,
)
from deerflow.sandbox.middleware import SandboxMiddleware, SandboxMiddlewareState
from deerflow.sandbox.prewarm import should_prewarm_sandbox
from deerflow.sandbox.sandbox import Sandbox
from deerflow.sandbox.sandbox_provider import SandboxProvider, reset_sandbox_provider, set_sandbox_provider
from deerflow.sandbox.search import GrepMatch
from deerflow.sandbox.tools import ensure_sandbox_initialized, ls_tool
from deerflow.sandbox.tools import ensure_sandbox_initialized, ensure_sandbox_initialized_async, ls_tool
class _SyncProvider(SandboxProvider):
@ -172,6 +176,24 @@ class _AsyncOnlyProvider(SandboxProvider):
return None
class _DelayedAsyncProvider(_AsyncOnlyProvider):
def __init__(self) -> None:
super().__init__()
self.started = asyncio.Event()
self.allow_finish = asyncio.Event()
async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
self.thread_ids.append(thread_id)
self.user_ids.append(user_id)
self.started.set()
await self.allow_finish.wait()
return "async-sandbox"
def _aio_prewarm_config() -> SimpleNamespace:
return SimpleNamespace(sandbox=SimpleNamespace(prewarm=True, use="deerflow.community.aio_sandbox:AioSandboxProvider"))
def test_sandbox_middleware_state_matches_thread_state_sandbox_field() -> None:
"""Middleware-local schema must not drift from ThreadState.sandbox."""
middleware_hints = get_type_hints(SandboxMiddlewareState, include_extras=True)
@ -442,6 +464,102 @@ async def test_default_lazy_tool_acquisition_uses_async_provider() -> None:
assert runtime.context["sandbox_id"] == "async-sandbox"
@pytest.mark.anyio
async def test_prewarm_overlaps_model_call_and_is_claimed_by_first_tool(monkeypatch: pytest.MonkeyPatch) -> None:
provider = _DelayedAsyncProvider()
monkeypatch.setattr("deerflow.sandbox.prewarm.get_app_config", _aio_prewarm_config)
set_sandbox_provider(provider)
context = {"thread_id": "thread-prewarm", "user_id": "owner-prewarm"}
ensure_sandbox_lease_owner(context)
runtime = Runtime(context=context)
tool_runtime = ToolRuntime(
state={},
context=context,
config={"configurable": {}},
stream_writer=lambda _: None,
tools=[],
tool_call_id="call-prewarm",
store=None,
)
try:
middleware = SandboxMiddleware()
register_sandbox_prewarm_cleanup(context)
await middleware.abefore_model({"messages": [HumanMessage(content="```sh\ngit status\n```")]}, runtime)
await asyncio.wait_for(provider.started.wait(), timeout=5)
tool_task = asyncio.create_task(ensure_sandbox_initialized_async(tool_runtime))
await asyncio.sleep(0)
assert provider.thread_ids == ["thread-prewarm"]
provider.allow_finish.set()
sandbox = await tool_task
finally:
provider.allow_finish.set()
await release_sandbox_execution_lease_async(context)
reset_sandbox_provider()
assert sandbox.id == "async-sandbox"
assert tool_runtime.state["sandbox"] == {"sandbox_id": "async-sandbox"}
assert provider.thread_ids == ["thread-prewarm"]
@pytest.mark.anyio
async def test_unused_prewarm_is_released_after_run(monkeypatch: pytest.MonkeyPatch) -> None:
provider = _AsyncOnlyProvider()
monkeypatch.setattr("deerflow.sandbox.prewarm.get_app_config", _aio_prewarm_config)
set_sandbox_provider(provider)
runtime = Runtime(context={"thread_id": "thread-unused", "user_id": "owner-unused"})
ensure_sandbox_lease_owner(runtime.context)
try:
register_sandbox_prewarm_cleanup(runtime.context)
middleware = SandboxMiddleware()
await middleware.abefore_model({"messages": [HumanMessage(content="Traceback (most recent call last):")]}, runtime)
await runtime.context["__deerflow_sandbox_prewarm"].task
await middleware.aafter_agent({}, runtime)
assert provider.released_ids == ["async-sandbox"]
finally:
await release_sandbox_execution_lease_async(runtime.context)
reset_sandbox_provider()
assert provider.thread_ids == ["thread-unused"]
assert provider.released_ids == ["async-sandbox"]
@pytest.mark.parametrize(
"content",
[
"Please explain this concept.",
"The source file is interesting but no path or command is supplied.",
],
)
@pytest.mark.anyio
async def test_prewarm_skips_messages_without_structured_signal(monkeypatch: pytest.MonkeyPatch, content: str) -> None:
provider = _AsyncOnlyProvider()
monkeypatch.setattr("deerflow.sandbox.prewarm.get_app_config", _aio_prewarm_config)
set_sandbox_provider(provider)
runtime = Runtime(context={"thread_id": "thread-no-prewarm"})
ensure_sandbox_lease_owner(runtime.context)
try:
register_sandbox_prewarm_cleanup(runtime.context)
await SandboxMiddleware().abefore_model({"messages": [HumanMessage(content=content)]}, runtime)
await asyncio.sleep(0)
finally:
await release_sandbox_execution_lease_async(runtime.context)
reset_sandbox_provider()
assert provider.thread_ids == []
def test_sandbox_config_prewarm_defaults_disabled_and_accepts_true() -> None:
from deerflow.config.sandbox_config import SandboxConfig
assert SandboxConfig(use="test").prewarm is False
assert SandboxConfig(use="test", prewarm=True).prewarm is True
@pytest.mark.parametrize("content", ["Review /mnt/user-data/workspace/app.py", r"Open C:\\work\\app.py"])
def test_prewarm_recognizes_unambiguous_unix_and_windows_paths(content: str) -> None:
assert should_prewarm_sandbox({"messages": [HumanMessage(content=content)]}) is True
@pytest.mark.anyio
@pytest.mark.parametrize(
("state", "runtime", "expected_sandbox_id"),

View File

@ -0,0 +1,793 @@
"""Lifecycle regressions for optional sandbox speculation."""
from __future__ import annotations
import asyncio
import logging
from types import SimpleNamespace
import pytest
from langchain_core.messages import HumanMessage
from deerflow.sandbox import prewarm
from deerflow.sandbox.exceptions import SandboxAuthorizationError
from deerflow.sandbox.lease import (
SANDBOX_PREWARM_CLEANUP_CONTEXT_KEY,
ensure_sandbox_lease_owner,
get_sandbox_lease_manager,
register_sandbox_prewarm_cleanup,
release_sandbox_execution_lease_async,
sandbox_prewarm_cleanup_registered,
)
from deerflow.sandbox.sandbox_provider import SandboxProvider, reset_sandbox_provider, set_sandbox_provider
from deerflow.sandbox.tools import ensure_sandbox_initialized_async
class Provider(SandboxProvider):
def __init__(self):
self.started = asyncio.Event()
self.finish = asyncio.Event()
self.finish.set()
self.calls = []
self.released = []
self.clients = {}
self.fail_once = False
def acquire(self, thread_id=None, *, user_id=None):
raise AssertionError("must use async provider hook")
async def acquire_async(self, thread_id=None, *, user_id=None):
self.calls.append((user_id, thread_id))
self.started.set()
await self.finish.wait()
if self.fail_once:
self.fail_once = False
raise RuntimeError("prewarm failed")
sid = f"{user_id}/{thread_id}"
self.clients[sid] = SimpleNamespace(id=sid, release_command_scope=lambda _: None)
return sid
def get(self, sandbox_id):
return self.clients.get(sandbox_id)
def get_scoped(self, sandbox_id, *, thread_id, user_id):
return self.get(sandbox_id) if sandbox_id == f"{user_id}/{thread_id}" else None
def release(self, sandbox_id):
self.released.append(sandbox_id)
self.clients.pop(sandbox_id, None)
def reset(self):
for sandbox_id in list(self.clients):
self.release(sandbox_id)
@pytest.fixture
def setup(monkeypatch):
provider = Provider()
config = SimpleNamespace(sandbox=SimpleNamespace(prewarm=True, prewarm_max_concurrent=1, use="deerflow.community.aio_sandbox:AioSandboxProvider"))
monkeypatch.setattr(prewarm, "get_app_config", lambda: config)
set_sandbox_provider(provider)
yield provider, config
reset_sandbox_provider()
def runtime(user="alice", thread="thread"):
context = {"user_id": user, "thread_id": thread}
ensure_sandbox_lease_owner(context)
register_sandbox_prewarm_cleanup(context)
return SimpleNamespace(context=context, state={})
def start(rt):
prewarm.start_sandbox_prewarm({"messages": [HumanMessage(content="```sh\npytest tests\n```")]}, rt)
return rt.context["__deerflow_sandbox_prewarm"]
@pytest.mark.parametrize("text", ["https://example.com/path/file.py", "https://example.com", "Python is useful", "git is a version control tool", "Please compare read/write behavior", "API/TUI", "Error: why is this phrasing confusing?"])
def test_non_execution_text_does_not_speculate(text):
assert not prewarm.should_prewarm_sandbox({"messages": [HumanMessage(content=text)]})
@pytest.mark.anyio
async def test_denied_prewarm_never_acquires(setup, monkeypatch, caplog):
provider, _ = setup
async def deny(**kwargs):
raise SandboxAuthorizationError()
monkeypatch.setattr(prewarm, "authorize_sandbox_execution_async", deny, raising=False)
rt = runtime()
with caplog.at_level(logging.DEBUG, logger=prewarm.__name__):
handle = start(rt)
await handle.task
await release_sandbox_execution_lease_async(rt.context)
assert provider.calls == []
records = [record for record in caplog.records if record.name == prewarm.__name__]
assert len(records) == 1
assert records[0].levelno == logging.DEBUG
assert records[0].exc_info is None
@pytest.mark.anyio
async def test_unexpected_prewarm_failure_retains_diagnostic_traceback(setup, caplog):
provider, _ = setup
provider.fail_once = True
rt = runtime()
with caplog.at_level(logging.INFO, logger=prewarm.__name__):
handle = start(rt)
await handle.task
await release_sandbox_execution_lease_async(rt.context)
records = [record for record in caplog.records if record.name == prewarm.__name__]
assert len(records) == 1
assert records[0].levelno == logging.INFO
assert records[0].exc_info[0] is RuntimeError
assert not prewarm._permits
@pytest.mark.anyio
async def test_unused_ready_prewarm_is_owned_and_outer_cleanup_releases(setup):
provider, _ = setup
rt = runtime()
handle = start(rt)
await handle.task
owner = rt.context["sandbox_lease_owner_id"]
assert get_sandbox_lease_manager(provider).binding_for(owner) == "alice/thread"
assert "sandbox_id" not in rt.context
await release_sandbox_execution_lease_async(rt.context)
await release_sandbox_execution_lease_async(rt.context)
assert provider.released == ["alice/thread"]
assert get_sandbox_lease_manager(provider).binding_for(owner) is None
@pytest.mark.anyio
async def test_first_tool_claims_same_lease_after_auth_recheck(setup, monkeypatch):
provider, _ = setup
rt = runtime()
handle = start(rt)
await handle.task
async def deny(**kwargs):
raise SandboxAuthorizationError()
monkeypatch.setattr("deerflow.sandbox.tools.authorize_sandbox_execution_async", deny)
with pytest.raises(SandboxAuthorizationError):
await ensure_sandbox_initialized_async(rt)
assert rt.state == {}
await release_sandbox_execution_lease_async(rt.context)
assert provider.released == ["alice/thread"]
@pytest.mark.anyio
async def test_ready_unused_counts_against_budget(setup):
provider, _ = setup
first, second = runtime(), runtime("bob")
await start(first).task
await start(second).task
assert provider.calls == [("alice", "thread")]
await release_sandbox_execution_lease_async(first.context)
third = runtime("carol")
await start(third).task
assert provider.calls[-1] == ("carol", "thread")
await release_sandbox_execution_lease_async(second.context)
await release_sandbox_execution_lease_async(third.context)
@pytest.mark.anyio
async def test_cancelled_claim_drains_late_acquire_through_repeated_cancellation(setup):
provider, _ = setup
provider.finish.clear()
rt = runtime()
handle = start(rt)
await provider.started.wait()
claim = asyncio.create_task(ensure_sandbox_initialized_async(rt))
await asyncio.sleep(0)
claim.cancel()
with pytest.raises(asyncio.CancelledError):
await claim
cleanup = asyncio.create_task(release_sandbox_execution_lease_async(rt.context))
await asyncio.sleep(0)
cleanup.cancel()
await asyncio.sleep(0)
cleanup.cancel()
await asyncio.sleep(0)
assert not cleanup.done()
provider.finish.set()
with pytest.raises(asyncio.CancelledError):
await cleanup
assert handle.task.done()
assert provider.released == ["alice/thread"]
assert get_sandbox_lease_manager(provider).binding_for(rt.context["sandbox_lease_owner_id"]) is None
@pytest.mark.anyio
async def test_unused_prewarm_does_not_release_sibling_holder(setup):
provider, _ = setup
rt = runtime()
await start(rt).task
manager = get_sandbox_lease_manager(provider)
await manager.reuse_or_acquire_async("sibling", "alice/thread", thread_id="thread", user_id="alice")
await release_sandbox_execution_lease_async(rt.context)
assert provider.released == []
assert provider.get("alice/thread") is not None
await manager.release_async("sibling")
assert provider.released == ["alice/thread"]
@pytest.mark.anyio
async def test_tool_hit_and_failed_prewarm_fallback_keep_one_owner(setup):
provider, _ = setup
for failing in (False, True):
provider.fail_once = failing
rt = runtime(thread=f"failure-{failing}")
handle = start(rt)
await handle.task
client = await ensure_sandbox_initialized_async(rt)
assert client.id == f"alice/failure-{failing}"
assert get_sandbox_lease_manager(provider).binding_for(rt.context["sandbox_lease_owner_id"]) == client.id
await release_sandbox_execution_lease_async(rt.context)
assert len(provider.calls) == 3
assert len(provider.released) == 2
@pytest.mark.anyio
async def test_replaced_provider_is_not_claimed_or_released_by_wrong_manager(setup):
old, _ = setup
rt = runtime()
handle = start(rt)
await handle.task
replacement = Provider()
reset_sandbox_provider()
set_sandbox_provider(replacement)
# Reset owns destruction of old clients; a late claim must neither reuse
# that detached provider nor release the new provider through its old handle.
client = await ensure_sandbox_initialized_async(rt)
assert client is replacement.get("alice/thread")
assert old.released == ["alice/thread"]
assert replacement.released == []
await release_sandbox_execution_lease_async(rt.context)
assert replacement.released == ["alice/thread"]
@pytest.mark.anyio
async def test_budget_pending_and_claim_releases_slot_for_another_user(setup):
provider, _ = setup
provider.finish.clear()
first = runtime()
handle = start(first)
await provider.started.wait()
second = runtime("bob")
await start(second).task
assert provider.calls == [("alice", "thread")]
provider.finish.set()
await ensure_sandbox_initialized_async(first)
third = runtime("carol")
await start(third).task
assert provider.calls == [("alice", "thread"), ("carol", "thread")]
for rt in (first, second, third):
await release_sandbox_execution_lease_async(rt.context)
assert handle.claimed
@pytest.mark.anyio
@pytest.mark.parametrize("mode", ["disabled", "other-provider", "snapshot-disabled", "existing", "fork"])
async def test_prewarm_admission_preserves_existing_modes(setup, mode):
from langgraph.types import Overwrite
provider, config = setup
rt = runtime()
state = {"messages": [HumanMessage(content="```sh\npytest\n```")]}
if mode == "disabled":
config.sandbox.prewarm = False
elif mode == "other-provider":
config.sandbox.use = "deerflow.sandbox.local:LocalSandboxProvider"
elif mode == "snapshot-disabled":
rt.context["app_config"] = SimpleNamespace(sandbox=SimpleNamespace(prewarm=False))
elif mode == "existing":
state["sandbox"] = {"sandbox_id": "existing"}
else:
state["sandbox"] = Overwrite({"sandbox_id": "parent"})
prewarm.start_sandbox_prewarm(state, rt)
handle = rt.context.get("__deerflow_sandbox_prewarm")
if handle is not None:
await handle.task
await release_sandbox_execution_lease_async(rt.context)
assert provider.calls == []
@pytest.mark.anyio
@pytest.mark.parametrize("failure", [False, True])
async def test_compiled_graph_model_outcome_cleans_unclaimed_prewarm(setup, monkeypatch, failure):
from langchain.agents import create_agent
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from deerflow.sandbox.middleware import SandboxMiddleware
provider, _ = setup
class Model(BaseChatModel):
@property
def _llm_type(self):
return "prewarm-lifecycle-test"
def bind_tools(self, tools, **kwargs):
return self
def _generate(self, messages, stop=None, run_manager=None, **kwargs):
raise AssertionError("async model required")
async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs):
await provider.started.wait()
if failure:
raise RuntimeError("model failed before tool call")
return ChatResult(generations=[ChatGeneration(message=AIMessage(content="No execution needed"))])
middleware = SandboxMiddleware()
monkeypatch.setattr(middleware, "_prepare_agent_skill_projection", lambda *args, **kwargs: None)
graph = create_agent(Model(), tools=[], middleware=[middleware])
rt = runtime()
try:
if failure:
with pytest.raises(RuntimeError, match="model failed"):
await graph.ainvoke({"messages": [HumanMessage(content="```sh\npytest\n```")]}, context=rt.context)
else:
result = await graph.ainvoke({"messages": [HumanMessage(content="```sh\npytest\n```")]}, context=rt.context)
assert result["messages"][-1].content == "No execution needed"
finally:
await release_sandbox_execution_lease_async(rt.context)
assert provider.released == ["alice/thread"]
assert get_sandbox_lease_manager(provider).binding_for(rt.context["sandbox_lease_owner_id"]) is None
@pytest.mark.anyio
@pytest.mark.parametrize("cancelled", [False, True], ids=["model-error", "cancel"])
async def test_gateway_worker_finally_cleans_prewarm_without_context_sandbox_id(setup, cancelled):
from unittest.mock import AsyncMock
from deerflow.runtime.runs.manager import RunManager
from deerflow.runtime.runs.worker import RunContext, run_agent
provider, _ = setup
manager = RunManager()
record = await manager.create("gateway-prewarm-error")
contexts = []
ready = asyncio.Event()
blocked = asyncio.Event()
class FailingAgent:
async def astream(self, graph_input, config=None, **kwargs):
context = config["configurable"]["__pregel_runtime"].context
contexts.append(context)
ensure_sandbox_lease_owner(context)
await start(SimpleNamespace(context=context)).task
assert "sandbox_id" not in context
ready.set()
if cancelled:
await blocked.wait()
raise RuntimeError("failure after speculative startup")
yield
bridge = SimpleNamespace(publish=AsyncMock(), publish_end=AsyncMock(), cleanup=AsyncMock())
task = asyncio.create_task(run_agent(bridge, manager, record, ctx=RunContext(checkpointer=None), agent_factory=lambda **kwargs: FailingAgent(), graph_input={}, config={}))
try:
await asyncio.wait_for(ready.wait(), timeout=5)
if cancelled:
task.cancel()
await task
finally:
if not task.done():
task.cancel()
await asyncio.gather(task, return_exceptions=True)
assert len(provider.calls) == len(provider.released) == 1
assert not prewarm._permits
assert not sandbox_prewarm_cleanup_registered(contexts[0])
@pytest.mark.anyio
async def test_config_and_cold_provider_setup_do_not_block_event_loop(setup, monkeypatch):
import threading
provider, config = setup
loop_thread = threading.get_ident()
seen = []
def load_config():
assert threading.get_ident() != loop_thread
seen.append("config")
return config
def load_provider():
assert threading.get_ident() != loop_thread
seen.append("provider")
return provider
monkeypatch.setattr(prewarm, "get_app_config", load_config)
monkeypatch.setattr(prewarm, "get_initialized_sandbox_provider", lambda: None)
monkeypatch.setattr(prewarm, "get_sandbox_provider", load_provider)
rt = runtime()
await start(rt).task
await release_sandbox_execution_lease_async(rt.context)
assert seen == ["config", "provider"]
@pytest.mark.anyio
async def test_concurrent_cleanup_waits_for_same_pending_reconciliation(setup):
provider, _ = setup
provider.finish.clear()
rt = runtime()
handle = start(rt)
await provider.started.wait()
first = asyncio.create_task(release_sandbox_execution_lease_async(rt.context))
second = asyncio.create_task(release_sandbox_execution_lease_async(rt.context))
await asyncio.sleep(0)
assert not first.done() and not second.done()
provider.finish.set()
await asyncio.gather(first, second)
assert provider.released == ["alice/thread"]
assert handle.closed and handle.task.done()
assert not prewarm._permits
@pytest.mark.anyio
async def test_no_second_speculation_in_same_execution_after_no_tool_turn(setup):
provider, _ = setup
rt = runtime()
first = start(rt)
await first.task
await release_sandbox_execution_lease_async(rt.context)
assert start(rt) is first
assert provider.calls == [("alice", "thread")]
@pytest.mark.anyio
async def test_speculative_budget_is_shared_across_isolated_event_loops(setup):
import threading
provider, _ = setup
admitted = threading.Barrier(5)
finish = threading.Barrier(5)
def run_user(user):
async def run():
rt = runtime(user)
await start(rt).task
await asyncio.to_thread(admitted.wait, 5)
await asyncio.to_thread(finish.wait, 5)
await release_sandbox_execution_lease_async(rt.context)
asyncio.run(run())
tasks = [asyncio.create_task(asyncio.to_thread(run_user, f"user-{i}")) for i in range(4)]
try:
await asyncio.to_thread(admitted.wait, 5)
assert len(provider.calls) == 1
finally:
await asyncio.to_thread(finish.wait, 5)
await asyncio.gather(*tasks)
assert len(provider.released) == 1
assert not prewarm._permits
@pytest.mark.anyio
async def test_repeated_cancel_during_release_keeps_budget_until_cleanup_finishes(setup, monkeypatch):
import threading
provider, _ = setup
rt = runtime()
await start(rt).task
started, finish = threading.Event(), threading.Event()
release = provider.release
def blocking_release(sandbox_id):
started.set()
assert finish.wait(5)
release(sandbox_id)
monkeypatch.setattr(provider, "release", blocking_release)
cleanup = asyncio.create_task(release_sandbox_execution_lease_async(rt.context))
try:
assert await asyncio.to_thread(started.wait, 5)
cleanup.cancel()
await asyncio.sleep(0)
cleanup.cancel()
await asyncio.sleep(0)
assert not cleanup.done()
other = runtime("bob")
await start(other).task
assert provider.calls == [("alice", "thread")]
await release_sandbox_execution_lease_async(other.context)
finally:
finish.set()
with pytest.raises(asyncio.CancelledError):
await cleanup
assert provider.released == ["alice/thread"]
assert not prewarm._permits
@pytest.mark.anyio
async def test_cancel_before_start_creates_no_provider_lease(setup):
provider, _ = setup
rt = runtime()
handle = start(rt)
handle.task.cancel()
await release_sandbox_execution_lease_async(rt.context)
assert provider.calls == []
assert not prewarm._permits
@pytest.mark.anyio
@pytest.mark.parametrize("mode", ["disabled", "non-aio", "budget-full"])
async def test_cancelled_skipped_prewarm_cleanup_still_releases_actual_tool_lease(setup, monkeypatch, mode):
provider, config = setup
blocker = None
if mode == "disabled":
config.sandbox.prewarm = False
elif mode == "non-aio":
config.sandbox.use = "deerflow.sandbox.local:LocalSandboxProvider"
else:
blocker = runtime("slot-owner")
await start(blocker).task
rt = runtime()
handle = start(rt)
assert await handle.task is None
await ensure_sandbox_initialized_async(rt)
owner = rt.context["sandbox_lease_owner_id"]
manager = get_sandbox_lease_manager(provider)
assert manager.binding_for(owner) == "alice/thread"
entered, finish = asyncio.Event(), asyncio.Event()
wait_for_cleanup = handle._wait_for_cleanup
async def gated_wait():
entered.set()
await finish.wait()
await wait_for_cleanup()
monkeypatch.setattr(handle, "_wait_for_cleanup", gated_wait)
cleanup = asyncio.create_task(release_sandbox_execution_lease_async(rt.context))
try:
await entered.wait()
cleanup.cancel()
await asyncio.sleep(0)
cleanup.cancel()
finish.set()
with pytest.raises(asyncio.CancelledError):
await cleanup
assert manager.binding_for(owner) is None
assert "alice/thread" in provider.released
finally:
finish.set()
await asyncio.gather(cleanup, return_exceptions=True)
await manager.release_async(owner)
if blocker is not None:
await release_sandbox_execution_lease_async(blocker.context)
@pytest.mark.anyio
async def test_prewarm_close_error_still_releases_actual_tool_lease(setup, monkeypatch):
provider, config = setup
config.sandbox.prewarm = False
rt = runtime()
await start(rt).task
await ensure_sandbox_initialized_async(rt)
manager = get_sandbox_lease_manager(provider)
owner = rt.context["sandbox_lease_owner_id"]
async def fail_close(context):
raise RuntimeError("prewarm cleanup failed")
monkeypatch.setattr(prewarm, "close_sandbox_prewarm_context", fail_close)
try:
with pytest.raises(RuntimeError, match="prewarm cleanup failed"):
await release_sandbox_execution_lease_async(rt.context)
assert manager.binding_for(owner) is None
assert provider.released == ["alice/thread"]
finally:
await manager.release_async(owner)
@pytest.mark.anyio
async def test_cancel_after_disabled_prewarm_close_drains_blocked_ordinary_release(setup, monkeypatch):
import threading
provider, config = setup
config.sandbox.prewarm = False
rt = runtime()
handle = start(rt)
await handle.task
await ensure_sandbox_initialized_async(rt)
close_started, finish_close = asyncio.Event(), asyncio.Event()
release_started, finish_release = threading.Event(), threading.Event()
original_wait = handle._wait_for_cleanup
original_release = provider.release
async def gated_wait():
close_started.set()
await finish_close.wait()
await original_wait()
def blocking_release(sandbox_id):
release_started.set()
assert finish_release.wait(5)
original_release(sandbox_id)
monkeypatch.setattr(handle, "_wait_for_cleanup", gated_wait)
monkeypatch.setattr(provider, "release", blocking_release)
cleanup = asyncio.create_task(release_sandbox_execution_lease_async(rt.context))
try:
await close_started.wait()
cleanup.cancel()
finish_close.set()
assert await asyncio.to_thread(release_started.wait, 1)
cleanup.cancel()
await asyncio.sleep(0)
cleanup.cancel()
await asyncio.sleep(0)
assert not cleanup.done()
finish_release.set()
with pytest.raises(asyncio.CancelledError):
await cleanup
assert provider.released == ["alice/thread"]
finally:
finish_close.set()
finish_release.set()
await asyncio.gather(cleanup, return_exceptions=True)
await get_sandbox_lease_manager(provider).release_async(rt.context["sandbox_lease_owner_id"])
@pytest.mark.parametrize("enabled", [False, True])
@pytest.mark.parametrize("method", ["invoke", "stream"])
def test_compiled_sync_graph_keeps_lazy_path_without_speculation(setup, monkeypatch, enabled, method):
from langchain.agents import create_agent
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from deerflow.sandbox.middleware import SandboxMiddleware
provider, config = setup
config.sandbox.prewarm = enabled
class Model(BaseChatModel):
@property
def _llm_type(self):
return "sync-prewarm-compatibility-test"
def bind_tools(self, tools, **kwargs):
return self
def _generate(self, messages, stop=None, run_manager=None, **kwargs):
return ChatResult(generations=[ChatGeneration(message=AIMessage(content="Explained without execution"))])
middleware = SandboxMiddleware()
monkeypatch.setattr(middleware, "_prepare_agent_skill_projection", lambda *args, **kwargs: None)
graph = create_agent(Model(), tools=[], middleware=[middleware])
context = {"thread_id": "sync-thread", "user_id": "alice", "app_config": config}
state = {"messages": [HumanMessage(content="```python\nprint(42)\n```")]}
if method == "invoke":
result = graph.invoke(state, context=context)
else:
result = list(graph.stream(state, context=context, stream_mode="values"))[-1]
assert result["messages"][-1].content == "Explained without execution"
assert "__deerflow_sandbox_prewarm" not in context
assert provider.calls == []
assert not prewarm._permits
@pytest.mark.anyio
@pytest.mark.parametrize("enabled", [False, True])
@pytest.mark.parametrize("outcome", ["success", "model-error", "cancel", "tool"])
@pytest.mark.parametrize("method", ["ainvoke", "astream"])
async def test_direct_sdk_graph_does_not_leak_speculation(setup, monkeypatch, enabled, outcome, method):
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from langchain_core.tools import StructuredTool
from deerflow.agents.factory import create_deerflow_agent
from deerflow.agents.features import RuntimeFeatures
from deerflow.sandbox.middleware import SandboxMiddleware
provider, config = setup
config.sandbox.prewarm = enabled
context = {"thread_id": "direct-graph", "user_id": "alice", "app_config": config}
entered = asyncio.Event()
blocked = asyncio.Event()
class Model(BaseChatModel):
@property
def _llm_type(self):
return "direct-sdk-prewarm-test"
def bind_tools(self, tools, **kwargs):
return self
def _generate(self, *args, **kwargs):
raise AssertionError("async model required")
async def _agenerate(self, *args, **kwargs):
messages = args[0] if args else kwargs["messages"]
handle = context.get("__deerflow_sandbox_prewarm")
if handle is not None:
await handle.task
entered.set()
if outcome == "model-error":
raise RuntimeError("direct model failed")
if outcome == "cancel":
await blocked.wait()
if outcome == "tool" and not any(isinstance(message, ToolMessage) for message in messages):
return ChatResult(generations=[ChatGeneration(message=AIMessage(content="", tool_calls=[{"name": "use_sandbox", "args": {}, "id": "sandbox-call"}]))])
return ChatResult(generations=[ChatGeneration(message=AIMessage(content="No execution needed"))])
monkeypatch.setattr(SandboxMiddleware, "_prepare_agent_skill_projection", lambda *args, **kwargs: None)
async def use_sandbox():
sandbox = await ensure_sandbox_initialized_async(SimpleNamespace(context=context, state={}))
return sandbox.id
tool = StructuredTool.from_function(coroutine=use_sandbox, name="use_sandbox", description="Acquire a sandbox for a real tool call")
graph = create_deerflow_agent(Model(), tools=[tool] if outcome == "tool" else [], features=RuntimeFeatures(loop_detection=False))
state = {"messages": [HumanMessage(content="```sh\npytest tests\n```")]}
async def invoke():
if method == "ainvoke":
return await graph.ainvoke(state, context=context)
return [chunk async for chunk in graph.astream(state, context=context, stream_mode="values")][-1]
task = asyncio.create_task(invoke())
try:
await asyncio.wait_for(entered.wait(), timeout=5)
if outcome == "cancel":
task.cancel()
with pytest.raises(asyncio.CancelledError):
await task
elif outcome == "model-error":
with pytest.raises(RuntimeError, match="direct model failed"):
await task
else:
result = await task
assert result["messages"][-1].content == "No execution needed"
# Observe the graph's own terminal state, before test teardown.
assert provider.clients == {}
assert get_sandbox_lease_manager(provider).binding_for(context["sandbox_lease_owner_id"]) is None
assert not prewarm._permits
if outcome == "tool":
assert provider.calls == [("alice", "direct-graph")]
assert provider.released == ["alice/direct-graph"]
finally:
if not task.done():
task.cancel()
await asyncio.gather(task, return_exceptions=True)
# Failure-only hygiene does not supply the graph's execution boundary.
handle = context.get("__deerflow_sandbox_prewarm")
if handle is not None:
await handle.close()
@pytest.mark.anyio
@pytest.mark.parametrize("marker", ["missing", None, True, "registered", {}])
async def test_prewarm_requires_registered_execution_cleanup(setup, marker):
provider, _ = setup
rt = runtime()
rt.context.pop(SANDBOX_PREWARM_CLEANUP_CONTEXT_KEY)
if marker != "missing":
rt.context[SANDBOX_PREWARM_CLEANUP_CONTEXT_KEY] = marker
prewarm.start_sandbox_prewarm({"messages": [HumanMessage(content="```sh\npytest tests\n```")]}, rt)
assert "__deerflow_sandbox_prewarm" not in rt.context
assert provider.calls == []
assert not prewarm._permits
@pytest.mark.anyio
async def test_execution_cleanup_revokes_prewarm_admission(setup):
rt = runtime()
assert sandbox_prewarm_cleanup_registered(rt.context)
await release_sandbox_execution_lease_async(rt.context)
assert not sandbox_prewarm_cleanup_registered(rt.context)

View File

@ -0,0 +1,24 @@
"""Operator controls bound speculative sandbox work without enabling it by default."""
import pytest
from pydantic import ValidationError
from deerflow.config.sandbox_config import SandboxConfig
def test_prewarm_defaults_are_opt_in_and_bounded():
config = SandboxConfig(use="deerflow.community.aio_sandbox:AioSandboxProvider")
assert config.prewarm is False
assert config.prewarm_max_concurrent == 1
@pytest.mark.parametrize("value", [1, 4, "3"])
def test_prewarm_limit_accepts_positive_integers_and_environment_strings(value):
config = SandboxConfig(use="example:Provider", prewarm_max_concurrent=value)
assert config.prewarm_max_concurrent == int(value)
@pytest.mark.parametrize("value", [True, False, 0, -1, 1.5, "bad", None])
def test_prewarm_limit_rejects_invalid_values(value):
with pytest.raises(ValidationError):
SandboxConfig(use="example:Provider", prewarm_max_concurrent=value)

View File

@ -2352,6 +2352,7 @@ class TestAsyncExecutionPath:
mock_agent.astream = failing_stream
sys.modules["deerflow.sandbox"].get_sandbox_provider.return_value = provider
monkeypatch.setattr(importlib.import_module("deerflow.sandbox.sandbox_provider"), "get_sandbox_provider", lambda: provider)
lease_module = importlib.import_module("deerflow.sandbox.lease")
monkeypatch.setattr(lease_module, "get_sandbox_lease_manager", lambda _provider: manager)
@ -2412,6 +2413,7 @@ class TestAsyncExecutionPath:
mock_agent.astream = failing_stream
sys.modules["deerflow.sandbox"].get_sandbox_provider.return_value = provider
monkeypatch.setattr(importlib.import_module("deerflow.sandbox.sandbox_provider"), "get_sandbox_provider", lambda: provider)
lease_module = importlib.import_module("deerflow.sandbox.lease")
monkeypatch.setattr(lease_module, "get_sandbox_lease_manager", lambda _provider: manager)
@ -3986,6 +3988,7 @@ class TestCooperativeCancellation:
lambda: object(),
raising=False,
)
monkeypatch.setattr(importlib.import_module("deerflow.sandbox.sandbox_provider"), "get_sandbox_provider", lambda: object())
lease_module = importlib.import_module("deerflow.sandbox.lease")
monkeypatch.setattr(
lease_module,
@ -6724,3 +6727,42 @@ def test_utcnow_helper_returns_utc_aware_datetime(classes):
assert now.tzinfo is not None
assert now.utcoffset() is not None
assert now.utcoffset().total_seconds() == 0.0
@pytest.mark.anyio
async def test_subagent_model_failure_drains_unclaimed_prewarm(classes, base_config, monkeypatch):
from deerflow.sandbox.lease import discard_sandbox_lease_manager, get_sandbox_lease_manager
prewarm = importlib.import_module("deerflow.sandbox.prewarm")
sandbox = SimpleNamespace(release_command_scope=lambda _: None)
provider = SimpleNamespace(
acquire_async=AsyncMock(return_value="prewarmed"),
get=lambda _: sandbox,
get_scoped=lambda *args, **kwargs: sandbox,
release=MagicMock(),
)
config = SimpleNamespace(sandbox=SimpleNamespace(prewarm=True, prewarm_max_concurrent=1, use="deerflow.community.aio_sandbox:AioSandboxProvider"))
monkeypatch.setattr(prewarm, "get_app_config", lambda: config)
monkeypatch.setattr(prewarm, "get_initialized_sandbox_provider", lambda: provider)
monkeypatch.setattr(prewarm, "authorize_sandbox_execution_async", AsyncMock())
captured = []
async def failing_stream(state, *, context, **kwargs):
captured.append(context)
prewarm.start_sandbox_prewarm({"messages": [classes["HumanMessage"](content="```sh\npytest\n```")]}, SimpleNamespace(context=context))
await context["__deerflow_sandbox_prewarm"].task
assert "sandbox_id" not in context
raise RuntimeError("subagent model failed after prewarm")
yield
agent = MagicMock()
agent.astream = failing_stream
executor = classes["SubagentExecutor"](config=base_config, tools=[], thread_id="subagent-prewarm", user_id="alice")
try:
with patch.object(executor, "_create_agent", return_value=agent):
result = await executor._aexecute("Task")
assert result.status == classes["SubagentStatus"].FAILED
provider.release.assert_called_once_with("prewarmed")
assert get_sandbox_lease_manager(provider).binding_for(captured[0]["sandbox_lease_owner_id"]) is None
finally:
discard_sandbox_lease_manager(provider)

View File

@ -20,7 +20,7 @@
# ============================================================================
# Bump this number when the config schema changes.
# Run `make config-upgrade` to merge new fields into your local config.yaml.
config_version: 60
config_version: 61
# Optional operator-owned literal prompt extensions (see CONFIGURATION.md):
# lead_prompt_overlay:
@ -1703,6 +1703,13 @@ sandbox:
# not a secure isolation boundary for shell access. Enable only for fully
# trusted, single-user local workflows.
allow_host_bash: false
# For AioSandboxProvider only: overlap likely sandbox work with the first
# model call. It is disabled by default and ignored by other providers.
prewarm: false
# Bound pending/ready-unused speculation per provider instance, per process.
# A full budget skips prewarming; foreground tools still acquire normally.
# This is not a total container or deployment-wide capacity limit.
prewarm_max_concurrent: 1
# Optional: Mount additional host directories into the sandbox.
# Each mount maps a host path to a virtual container path accessible by the agent.
# Note: with LocalSandboxProvider under `make up` (docker-compose), host_path is

View File

@ -135,7 +135,7 @@ they resolve from the `secrets` map):
```yaml
config: |
config_version: 60
config_version: 61
models:
- name: gpt-4
use: langchain_openai:ChatOpenAI

View File

@ -367,7 +367,7 @@ ingress:
# -- DeerFlow config.yaml content. Secrets MUST stay as $VAR references — never
# inline literal secret values here. The default enables provisioner sandbox.
config: |
config_version: 60
config_version: 61
log_level: info
recursion_limit: 100
max_recursion_limit: 1000
@ -393,6 +393,9 @@ config: |
sandbox:
use: deerflow.community.aio_sandbox:AioSandboxProvider
prewarm: false
# Pending/ready-unused speculation per provider instance in each process.
prewarm_max_concurrent: 1
provisioner_url: http://provisioner:8002
# Sent as X-API-Key on every provisioner request. The chart injects the
# generated `<release>-app` key into the gateway and provisioner Pods as

View File

@ -1,6 +1,6 @@
# AGENTS.md
This is the source of truth for DeerFlow frontend coding agents; the sibling `CLAUDE.md` imports it via `@AGENTS.md`.
DeerFlow frontend guidance source of truth; `CLAUDE.md` imports `@AGENTS.md`.
## Project Overview