from __future__ import annotations import asyncio import base64 import io import json import tarfile import uuid from pathlib import Path from typing import Any, cast import aiohttp import pytest from agents.extensions.sandbox.cloudflare import ( CloudflareBucketMountStrategy, CloudflareSandboxClient, CloudflareSandboxClientOptions, CloudflareSandboxSession, CloudflareSandboxSessionState, ) from agents.extensions.sandbox.cloudflare.sandbox import _CloudflarePtyProcessEntry from agents.sandbox.entries import Dir, GCSMount, R2Mount, S3Mount from agents.sandbox.errors import ( ConfigurationError, ErrorCode, ExecTimeoutError, ExecTransportError, InvalidManifestPathError, MountConfigError, PtySessionNotFoundError, WorkspaceArchiveReadError, WorkspaceArchiveWriteError, WorkspaceReadNotFoundError, WorkspaceStartError, WorkspaceWriteTypeError, ) from agents.sandbox.manifest import Environment, Manifest from agents.sandbox.session.dependencies import Dependencies from agents.sandbox.session.pty_types import PTY_PROCESSES_MAX, allocate_pty_process_id from agents.sandbox.snapshot import NoopSnapshot, SnapshotBase from agents.sandbox.types import ExecResult from agents.sandbox.workspace_paths import SandboxPathGrant _WORKER_URL = "https://sandbox-cf.example.workers.dev" class _FakeResponse: def __init__(self, status: int = 200, json_body: Any = None, raw_body: bytes = b"") -> None: self.status = status self._json_body = json_body self._raw_body = raw_body async def json(self, *, content_type: str | None = None) -> Any: _ = content_type if self._json_body is not None: return self._json_body return json.loads(self._raw_body) async def read(self) -> bytes: if self._json_body is not None: return json.dumps(self._json_body).encode() return self._raw_body async def __aenter__(self) -> _FakeResponse: return self async def __aexit__(self, *args: object) -> None: _ = args class _FakeStreamContent: def __init__(self, data: bytes) -> None: self._data = data async def iter_any(self) -> Any: yield self._data class _FakeSSEResponse: def __init__(self, status: int, sse_body: bytes) -> None: self.status = status self.content = _FakeStreamContent(sse_body) async def json(self, *, content_type: str | None = None) -> Any: _ = content_type return {} async def __aenter__(self) -> _FakeSSEResponse: return self async def __aexit__(self, *args: object) -> None: _ = args class _FakeHttp: def __init__( self, responses: dict[str, _FakeResponse | _FakeSSEResponse] | None = None ) -> None: self._responses: dict[tuple[str, str], _FakeResponse | _FakeSSEResponse] = {} self.default_response: _FakeResponse | _FakeSSEResponse = _FakeResponse( status=200, json_body={"ok": True} ) self.calls: list[dict[str, Any]] = [] self.closed = False self.ws_connect_calls: list[dict[str, Any]] = [] self.fake_ws: _FakeWebSocket | None = None if responses: for key, val in responses.items(): method, _, suffix = key.partition(" ") self._responses[(method.upper(), suffix)] = val def _match(self, method: str, url: str) -> _FakeResponse | _FakeSSEResponse: for (m, suffix), resp in self._responses.items(): if m == method and suffix in url: return resp return self.default_response def _record(self, method: str, url: str, **kwargs: Any) -> _FakeResponse | _FakeSSEResponse: self.calls.append({"method": method, "url": url, **kwargs}) return self._match(method, url) def post(self, url: str, **kwargs: Any) -> _FakeResponse | _FakeSSEResponse: return self._record("POST", url, **kwargs) def get(self, url: str, **kwargs: Any) -> _FakeResponse | _FakeSSEResponse: return self._record("GET", url, **kwargs) def put(self, url: str, **kwargs: Any) -> _FakeResponse | _FakeSSEResponse: return self._record("PUT", url, **kwargs) def delete(self, url: str, **kwargs: Any) -> _FakeResponse | _FakeSSEResponse: return self._record("DELETE", url, **kwargs) async def ws_connect(self, url: str, **kwargs: Any) -> _FakeWebSocket: self.ws_connect_calls.append({"url": url, **kwargs}) if self.fake_ws is None: raise RuntimeError("fake_ws must be set before ws_connect") return self.fake_ws async def close(self) -> None: self.closed = True class _FakeWebSocket: def __init__(self, frames: list[aiohttp.WSMessage] | None = None) -> None: self.frames = list(frames or []) self.sent_bytes: list[bytes] = [] self.closed = False async def receive(self) -> aiohttp.WSMessage: if self.frames: return self.frames.pop(0) return aiohttp.WSMessage(aiohttp.WSMsgType.CLOSED, None, None) async def send_bytes(self, data: bytes) -> None: self.sent_bytes.append(data) async def close(self) -> None: self.closed = True class _BlockingFakeWebSocket(_FakeWebSocket): async def receive(self) -> aiohttp.WSMessage: if self.frames: return self.frames.pop(0) await asyncio.sleep(60.0) return aiohttp.WSMessage(aiohttp.WSMsgType.CLOSED, None, None) def _valid_tar_bytes() -> bytes: """Return a minimal valid tar archive for hydrate tests.""" buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w") as tar: info = tarfile.TarInfo(name="hello.txt") data = b"hello" info.size = len(data) tar.addfile(info, io.BytesIO(data)) return buf.getvalue() class _RestorableSnapshot(SnapshotBase): type: str = "test_restorable_snapshot" payload: bytes = b"" def __init__(self, **kwargs: object) -> None: if "payload" not in kwargs: kwargs["payload"] = _valid_tar_bytes() super().__init__(**kwargs) # type: ignore[arg-type] async def persist(self, data: io.IOBase, *, dependencies: Dependencies | None = None) -> None: _ = (data, dependencies) return None async def restore(self, *, dependencies: Dependencies | None = None) -> io.IOBase: _ = dependencies return io.BytesIO(self.payload) async def restorable(self, *, dependencies: Dependencies | None = None) -> bool: _ = dependencies return True def _make_state( *, worker_url: str = _WORKER_URL, sandbox_id: str = "abc123", manifest: Manifest | None = None, ) -> CloudflareSandboxSessionState: return CloudflareSandboxSessionState( session_id=uuid.uuid4(), manifest=manifest or Manifest(), snapshot=NoopSnapshot(id="snapshot"), worker_url=worker_url, sandbox_id=sandbox_id, ) def _make_session( *, state: CloudflareSandboxSessionState | None = None, fake_http: _FakeHttp | None = None, exec_timeout_s: float | None = None, request_timeout_s: float | None = None, ) -> CloudflareSandboxSession: sess = CloudflareSandboxSession( state=state or _make_state(), http=cast(Any, fake_http), exec_timeout_s=exec_timeout_s, request_timeout_s=request_timeout_s, ) # Override remote path normalization so tests do not need a live exec endpoint # for the runtime helper script. Dedicated tests verify the override is wired in. async def _sync_normalize(path: Path | str, *, for_write: bool = False) -> Path: return sess.normalize_path(path, for_write=for_write) sess._validate_path_access = _sync_normalize # type: ignore[method-assign] return sess def _build_sse_body(stdout: str = "", stderr: str = "", exit_code: int = 0) -> bytes: parts: list[str] = [] if stdout: parts.append(f"event: stdout\ndata: {base64.b64encode(stdout.encode()).decode()}\n\n") if stderr: parts.append(f"event: stderr\ndata: {base64.b64encode(stderr.encode()).decode()}\n\n") parts.append(f'event: exit\ndata: {{"exit_code": {exit_code}}}\n\n') return "".join(parts).encode("utf-8") def _exec_ok_response(stdout: str = "", stderr: str = "", exit_code: int = 0) -> _FakeSSEResponse: return _FakeSSEResponse( status=200, sse_body=_build_sse_body(stdout=stdout, stderr=stderr, exit_code=exit_code), ) def _streamed_payload_response(*, payload: bytes, is_binary: bool) -> _FakeResponse: chunk = base64.b64encode(payload).decode() if is_binary else payload.decode() body = ( f'data: {{"type":"metadata","isBinary":{str(is_binary).lower()}}}\n\n' f'data: {{"type":"chunk","data":"{chunk}"}}\n\n' 'data: {"type":"complete"}\n\n' ).encode() return _FakeResponse(status=200, raw_body=body) def _truncated_streamed_payload_response(*, payload: bytes, is_binary: bool) -> _FakeResponse: chunk = base64.b64encode(payload).decode() if is_binary else payload.decode() body = ( f'data: {{"type":"metadata","isBinary":{str(is_binary).lower()}}}\n\n' f'data: {{"type":"chunk","data":"{chunk}"}}\n\n' ).encode() return _FakeResponse(status=200, raw_body=body) def _ws_text_frame(payload: dict[str, object]) -> aiohttp.WSMessage: return aiohttp.WSMessage(aiohttp.WSMsgType.TEXT, json.dumps(payload), None) def _ws_binary_frame(payload: bytes) -> aiohttp.WSMessage: return aiohttp.WSMessage(aiohttp.WSMsgType.BINARY, payload, None) async def _register_pty_entry( session: CloudflareSandboxSession, *, ws: _FakeWebSocket, tty: bool, last_used: float = 0.0, ) -> int: pty_entry = _CloudflarePtyProcessEntry(ws=cast(Any, ws), tty=tty, last_used=last_used) async with session._pty_lock: process_id = allocate_pty_process_id(session._reserved_pty_process_ids) session._reserved_pty_process_ids.add(process_id) session._pty_processes[process_id] = pty_entry return process_id def test_cloudflare_bucket_mount_strategy_round_trips_through_manifest_parse() -> None: manifest = Manifest.model_validate( { "entries": { "remote": { "type": "s3_mount", "bucket": "bucket", "mount_strategy": {"type": "cloudflare_bucket_mount"}, } } } ) mount = manifest.entries["remote"] assert isinstance(mount, S3Mount) assert isinstance(mount.mount_strategy, CloudflareBucketMountStrategy) def test_cloudflare_bucket_mount_strategy_builds_s3_config() -> None: strategy = CloudflareBucketMountStrategy() mount = S3Mount( bucket="bucket", access_key_id="access-key", secret_access_key="secret-key", prefix="nested/prefix/", mount_strategy=strategy, read_only=False, ) config = strategy._build_cloudflare_bucket_mount_config(mount) # noqa: SLF001 assert config.bucket_name == "bucket" assert config.bucket_endpoint_url == "https://s3.amazonaws.com" assert config.provider == "s3" assert config.key_prefix == "/nested/prefix/" assert config.credentials == { "access_key_id": "access-key", "secret_access_key": "secret-key", } assert config.read_only is False def test_cloudflare_bucket_mount_strategy_builds_r2_config() -> None: strategy = CloudflareBucketMountStrategy() mount = R2Mount( bucket="bucket", account_id="abc123accountid", access_key_id="access-key", secret_access_key="secret-key", mount_strategy=strategy, ) config = strategy._build_cloudflare_bucket_mount_config(mount) # noqa: SLF001 assert config.bucket_name == "bucket" assert config.bucket_endpoint_url == "https://abc123accountid.r2.cloudflarestorage.com" assert config.provider == "r2" assert config.key_prefix is None assert config.credentials == { "access_key_id": "access-key", "secret_access_key": "secret-key", } assert config.read_only is True def test_cloudflare_bucket_mount_strategy_builds_gcs_hmac_config() -> None: strategy = CloudflareBucketMountStrategy() mount = GCSMount( bucket="bucket", access_id="access-id", secret_access_key="secret-key", prefix="nested/prefix/", mount_strategy=strategy, read_only=False, ) config = strategy._build_cloudflare_bucket_mount_config(mount) # noqa: SLF001 assert config.bucket_name == "bucket" assert config.bucket_endpoint_url == "https://storage.googleapis.com" assert config.provider == "gcs" assert config.key_prefix == "/nested/prefix/" assert config.credentials == { "access_key_id": "access-id", "secret_access_key": "secret-key", } assert config.read_only is False def test_cloudflare_bucket_mount_strategy_rejects_gcs_native_auth() -> None: with pytest.raises( MountConfigError, match="gcs cloudflare bucket mounts require access_id and secret_access_key", ): GCSMount( bucket="bucket", service_account_file="/data/config/gcs.json", mount_strategy=CloudflareBucketMountStrategy(), ) def test_cloudflare_bucket_mount_strategy_rejects_s3_session_token() -> None: with pytest.raises( MountConfigError, match="cloudflare bucket mounts do not support s3 session_token credentials", ): S3Mount( bucket="bucket", access_key_id="access-key", secret_access_key="secret-key", session_token="session-token", mount_strategy=CloudflareBucketMountStrategy(), ) @pytest.mark.asyncio async def test_cloudflare_create_uses_client_timeouts( monkeypatch: pytest.MonkeyPatch, ) -> None: async def _fake_request_sandbox_id( self: CloudflareSandboxClient, worker_url: str, api_key: str | None, **kwargs: object ) -> str: return "mfrggzdfmy2tqnrzgezdgnbv" monkeypatch.setattr(CloudflareSandboxClient, "_request_sandbox_id", _fake_request_sandbox_id) client = CloudflareSandboxClient(exec_timeout_s=10.0, request_timeout_s=60.0) session = await client.create( options=CloudflareSandboxClientOptions( worker_url=_WORKER_URL, ), snapshot=None, ) state = cast(CloudflareSandboxSessionState, session.state) assert state.worker_url == _WORKER_URL assert state.sandbox_id == "mfrggzdfmy2tqnrzgezdgnbv" # Timeouts should NOT be persisted in state. assert not hasattr(state, "exec_timeout_s") assert not hasattr(state, "request_timeout_s") # But the session instance should have them from the client, not from options. inner = cast(CloudflareSandboxSession, session._inner) assert inner._exec_timeout_s == 10.0 assert inner._request_timeout_s == 60.0 @pytest.mark.asyncio async def test_cloudflare_create_uses_injected_api_key_for_auth_header( monkeypatch: pytest.MonkeyPatch, ) -> None: created_headers: list[dict[str, str]] = [] async def _fake_request_sandbox_id( self: CloudflareSandboxClient, worker_url: str, api_key: str | None, **kwargs: object ) -> str: return "mfrggzdfmy2tqnrzgezdgnbv" monkeypatch.setattr(CloudflareSandboxClient, "_request_sandbox_id", _fake_request_sandbox_id) class _RecordingClientSession: def __init__(self, *, headers: dict[str, str] | None = None) -> None: self.headers = headers or {} self.closed = False created_headers.append(self.headers) async def close(self) -> None: self.closed = True monkeypatch.setenv("CLOUDFLARE_SANDBOX_API_KEY", "env-token") monkeypatch.setattr(aiohttp, "ClientSession", _RecordingClientSession) client = CloudflareSandboxClient() session = await client.create( options=CloudflareSandboxClientOptions( worker_url=_WORKER_URL, api_key="injected-token", ), snapshot=None, ) inner = cast(CloudflareSandboxSession, session._inner) inner._session() assert created_headers == [{"Authorization": "Bearer injected-token"}] await inner._close_http() @pytest.mark.asyncio async def test_cloudflare_create_rejects_non_workspace_root() -> None: client = CloudflareSandboxClient() with pytest.raises(ConfigurationError) as exc_info: await client.create( options=CloudflareSandboxClientOptions(worker_url=_WORKER_URL), manifest=Manifest(root="/tmp/app"), snapshot=None, ) assert exc_info.value.error_code is ErrorCode.SANDBOX_CONFIG_INVALID assert exc_info.value.context["manifest_root"] == "/tmp/app" @pytest.mark.asyncio async def test_cloudflare_create_calls_post_sandbox_for_id( monkeypatch: pytest.MonkeyPatch, ) -> None: """Verify that create() calls POST /sandbox and uses the returned ID.""" requested_urls: list[str] = [] async def _fake_request_sandbox_id( self: CloudflareSandboxClient, worker_url: str, api_key: str | None, **kwargs: object ) -> str: requested_urls.append(worker_url) return "server2generated3id4base32" monkeypatch.setattr(CloudflareSandboxClient, "_request_sandbox_id", _fake_request_sandbox_id) client = CloudflareSandboxClient() session = await client.create( options=CloudflareSandboxClientOptions(worker_url=_WORKER_URL), snapshot=None, ) state = cast(CloudflareSandboxSessionState, session.state) assert state.sandbox_id == "server2generated3id4base32" assert requested_urls == [_WORKER_URL] @pytest.mark.asyncio async def test_cloudflare_create_raises_on_post_sandbox_failure() -> None: """Verify that create() raises ConfigurationError when POST /sandbox fails.""" client = CloudflareSandboxClient() with pytest.raises(ConfigurationError) as exc_info: await client.create( options=CloudflareSandboxClientOptions( worker_url="https://unreachable.invalid", ), snapshot=None, ) assert exc_info.value.error_code is ErrorCode.SANDBOX_CONFIG_INVALID @pytest.mark.asyncio async def test_cloudflare_resume_uses_client_timeouts(monkeypatch: pytest.MonkeyPatch) -> None: async def _running(self: CloudflareSandboxSession) -> bool: _ = self return False monkeypatch.setattr(CloudflareSandboxSession, "running", _running) client = CloudflareSandboxClient(exec_timeout_s=11.0, request_timeout_s=77.0) state = _make_state() session = await client.resume(state) inner = cast(CloudflareSandboxSession, session._inner) assert session.state is state # Timeouts come from the client, not from state. assert inner._exec_timeout_s == 11.0 assert inner._request_timeout_s == 77.0 @pytest.mark.asyncio @pytest.mark.parametrize( ("is_running", "workspace_root_ready", "workspace_preserved", "workspace_reusable"), [ (False, False, False, False), (False, True, False, False), (True, False, True, False), (True, True, True, True), ], ) async def test_cloudflare_resume_sets_preserved_state_from_running( monkeypatch: pytest.MonkeyPatch, is_running: bool, workspace_root_ready: bool, workspace_preserved: bool, workspace_reusable: bool, ) -> None: running_calls: list[str] = [] async def _running(self: CloudflareSandboxSession) -> bool: running_calls.append(self.state.sandbox_id) return is_running monkeypatch.setattr(CloudflareSandboxSession, "running", _running) client = CloudflareSandboxClient() state = _make_state() state.workspace_root_ready = workspace_root_ready session = await client.resume(state) inner = cast(CloudflareSandboxSession, session._inner) assert running_calls == ["abc123"] assert inner._workspace_state_preserved_on_start() is workspace_preserved # noqa: SLF001 assert inner._system_state_preserved_on_start() is workspace_preserved # noqa: SLF001 assert inner._can_reuse_preserved_workspace_on_resume() is workspace_reusable # noqa: SLF001 assert state.workspace_root_ready is (workspace_root_ready and is_running) @pytest.mark.asyncio async def test_cloudflare_exec_decodes_sse_output() -> None: sess = _make_session( fake_http=_FakeHttp({"POST /exec": _exec_ok_response(stdout="hello\n", stderr="warn")}) ) result = await sess._exec_internal("echo", "hello", timeout=5.0) assert result.stdout == b"hello\n" assert result.stderr == b"warn" assert result.exit_code == 0 @pytest.mark.asyncio async def test_cloudflare_exec_applies_manifest_environment() -> None: fake_http = _FakeHttp({"POST /exec": _exec_ok_response(stdout="hello")}) sess = _make_session( state=_make_state(manifest=Manifest(environment=Environment(value={"A": "1", "B": "two"}))), fake_http=fake_http, ) result = await sess._exec_internal("printenv", "A", timeout=5.0) assert result.exit_code == 0 exec_calls = [call for call in fake_http.calls if call["method"] == "POST"] assert exec_calls[0]["json"]["argv"] == ["env", "A=1", "B=two", "printenv", "A"] @pytest.mark.asyncio async def test_cloudflare_exec_timeout_raises_exec_timeout_error() -> None: class _TimeoutHttp(_FakeHttp): def post(self, url: str, **kwargs: Any) -> Any: self._record("POST", url, **kwargs) raise asyncio.TimeoutError() with pytest.raises(ExecTimeoutError): await _make_session(fake_http=_TimeoutHttp())._exec_internal("sleep", "999", timeout=1.0) @pytest.mark.asyncio async def test_cloudflare_exec_non_200_includes_provider_error_details() -> None: sess = _make_session( fake_http=_FakeHttp( { "POST /exec": _FakeResponse( status=502, json_body={ "error": "pool error: Failed to start container", "code": "pool_error", }, ) } ) ) with pytest.raises(ExecTransportError) as exc_info: await sess._exec_internal("mkdir", "-p", "--", "/workspace", timeout=5.0) assert exc_info.value.context == { "command": ("mkdir", "-p", "--", "/workspace"), "command_str": "mkdir -p -- /workspace", "backend": "cloudflare", "http_status": 502, "provider_error": "pool_error: pool error: Failed to start container", } assert ( str(exc_info.value.__cause__) == "POST /exec failed: HTTP 502: pool_error: pool error: Failed to start container" ) assert ( str(exc_info.value) == "POST /exec failed: HTTP 502: pool_error: pool error: Failed to start container" ) assert exc_info.value.retryable is None @pytest.mark.asyncio @pytest.mark.parametrize( ("status", "expected_retryable"), [ (400, False), (500, False), (503, True), ], ) async def test_cloudflare_exec_retryability_follows_documented_status_semantics( status: int, expected_retryable: bool, ) -> None: sess = _make_session( fake_http=_FakeHttp( { "POST /exec": _FakeResponse( status=status, json_body={ "error": "cloudflare sandbox error", "code": "cloudflare_error", }, ) } ) ) with pytest.raises(ExecTransportError) as exc_info: await sess._exec_internal("mkdir", "-p", "--", "/workspace", timeout=5.0) assert exc_info.value.context["backend"] == "cloudflare" assert exc_info.value.context["http_status"] == status assert exc_info.value.context["provider_error"] == "cloudflare_error: cloudflare sandbox error" assert exc_info.value.retryable is expected_retryable @pytest.mark.parametrize( ("status", "expected_retryable"), [ (400, False), (500, False), (503, True), (418, None), ], ) def test_cloudflare_retryability_status_table(status: int, expected_retryable: bool | None) -> None: from agents.extensions.sandbox.cloudflare import sandbox as mod assert mod._cloudflare_retryability_for_status(status) is expected_retryable @pytest.mark.asyncio async def test_cloudflare_exec_non_sse_json_body_includes_provider_error_details() -> None: sess = _make_session( fake_http=_FakeHttp( { "POST /exec": _FakeSSEResponse( status=200, sse_body=( b'{"error":"pool error: Failed to start container","code":"pool_error"}' ), ) } ) ) with pytest.raises(ExecTransportError) as exc_info: await sess._exec_internal("mkdir", "-p", "--", "/workspace", timeout=5.0) assert exc_info.value.context["http_status"] == 200 assert ( exc_info.value.context["provider_error"] == "pool_error: pool error: Failed to start container" ) assert str(exc_info.value.__cause__) == ( "POST /exec returned non-SSE error body: pool_error: pool error: Failed to start container" ) assert str(exc_info.value) == ( "POST /exec returned non-SSE error body: pool_error: pool error: Failed to start container" ) @pytest.mark.asyncio async def test_cloudflare_prepare_workspace_preserves_exec_error_context() -> None: sess = _make_session( fake_http=_FakeHttp( { "POST /exec": _FakeResponse( status=502, json_body={ "error": "pool error: Failed to start container", "code": "pool_error", }, ) } ) ) with pytest.raises(WorkspaceStartError) as exc_info: await sess._prepare_backend_workspace() assert exc_info.value.context["backend"] == "cloudflare" assert exc_info.value.context["reason"] == "prepare_workspace_exec_failed" exec_context = exc_info.value.context["exec_error_context"] assert isinstance(exec_context, dict) assert exec_context["http_status"] == 502 assert exec_context["provider_error"] == "pool_error: pool error: Failed to start container" assert str(exc_info.value) == ( "failed to start session: " "POST /exec failed: HTTP 502: pool_error: pool error: Failed to start container" ) @pytest.mark.asyncio async def test_cloudflare_exec_client_error_includes_provider_context() -> None: class _FailingHttp(_FakeHttp): def post(self, url: str, **kwargs: Any) -> Any: self._record("POST", url, **kwargs) raise aiohttp.ClientError("connection reset") with pytest.raises(ExecTransportError) as exc_info: await _make_session(fake_http=_FailingHttp())._exec_internal("echo", "hello", timeout=1.0) assert str(exc_info.value) == ( "Cloudflare exec transport failed: ClientError: connection reset" ) assert exc_info.value.context["backend"] == "cloudflare" assert exc_info.value.context["operation"] == "exec" assert exc_info.value.context["provider_error"] == "ClientError: connection reset" @pytest.mark.asyncio async def test_cloudflare_exec_stream_without_exit_raises_transport_error() -> None: sess = _make_session( fake_http=_FakeHttp( { "POST /exec": _FakeSSEResponse( status=200, sse_body=b"event: stdout\ndata: aGVsbG8=\n\n" ) } ) ) with pytest.raises(ExecTransportError): await sess._exec_internal("echo", "hello", timeout=5.0) @pytest.mark.asyncio async def test_cloudflare_read_and_write_use_file_endpoints() -> None: fake_http = _FakeHttp( { "GET /file/": _FakeResponse(status=200, raw_body=b"file-content"), "PUT /file/": _FakeResponse(status=200, json_body={"ok": True}), } ) sess = _make_session(fake_http=fake_http) result = await sess.read(Path("/workspace/test.txt")) assert result.read() == b"file-content" await sess.write(Path("/workspace/out.txt"), io.BytesIO(b"data")) get_calls = [c for c in fake_http.calls if c["method"] == "GET"] put_calls = [c for c in fake_http.calls if c["method"] == "PUT"] assert "/file/workspace/test.txt" in get_calls[0]["url"] assert "/file/workspace/out.txt" in put_calls[0]["url"] @pytest.mark.asyncio async def test_cloudflare_mount_and_unmount_bucket_use_http_endpoints() -> None: fake_http = _FakeHttp( { "POST /mount": _FakeResponse(status=200, json_body={"ok": True}), "POST /unmount": _FakeResponse(status=200, json_body={"ok": True}), } ) sess = _make_session(fake_http=fake_http) await sess.mount_bucket( bucket="my-bucket", mount_path=Path("/workspace/data"), options={ "endpoint": "https://s3.amazonaws.com", "readOnly": True, }, ) await sess.unmount_bucket(Path("/workspace/data")) mount_call = next(c for c in fake_http.calls if "/mount" in c["url"]) unmount_call = next(c for c in fake_http.calls if "/unmount" in c["url"]) assert mount_call["json"] == { "bucket": "my-bucket", "mountPath": "/workspace/data", "options": { "endpoint": "https://s3.amazonaws.com", "readOnly": True, }, } assert unmount_call["json"] == {"mountPath": "/workspace/data"} @pytest.mark.asyncio async def test_cloudflare_mount_and_unmount_validate_path_access_for_write() -> None: fake_http = _FakeHttp( { "POST /mount": _FakeResponse(status=200, json_body={"ok": True}), "POST /unmount": _FakeResponse(status=200, json_body={"ok": True}), } ) sess = _make_session(fake_http=fake_http) calls: list[tuple[str, bool]] = [] async def _tracking_normalize(path: Path | str, *, for_write: bool = False) -> Path: calls.append((Path(path).as_posix(), for_write)) return sess.normalize_path(path, for_write=for_write) sess._validate_path_access = _tracking_normalize # type: ignore[method-assign] await sess.mount_bucket( bucket="my-bucket", mount_path=Path("/workspace/data"), options={ "endpoint": "https://s3.amazonaws.com", "readOnly": True, }, ) await sess.unmount_bucket(Path("/workspace/data")) assert calls == [ ("/workspace/data", True), ("/workspace/data", True), ] @pytest.mark.asyncio async def test_cloudflare_mount_rejects_read_only_extra_path_grant() -> None: fake_http = _FakeHttp({"POST /mount": _FakeResponse(status=200, json_body={"ok": True})}) sess = _make_session( state=_make_state( manifest=Manifest( extra_path_grants=(SandboxPathGrant(path="/tmp/protected", read_only=True),) ) ), fake_http=fake_http, ) with pytest.raises(WorkspaceArchiveWriteError) as exc_info: await sess.mount_bucket( bucket="my-bucket", mount_path=Path("/tmp/protected/data"), options={ "endpoint": "https://s3.amazonaws.com", "readOnly": True, }, ) assert fake_http.calls == [] assert str(exc_info.value) == "failed to write archive for path: /tmp/protected/data" assert exc_info.value.context == { "path": "/tmp/protected/data", "reason": "read_only_extra_path_grant", "grant_path": "/tmp/protected", } async def test_cloudflare_read_decodes_streamed_file_payload() -> None: sess = _make_session( fake_http=_FakeHttp( {"GET /file/": _streamed_payload_response(payload=b"file-content", is_binary=False)} ) ) result = await sess.read(Path("/workspace/test.txt")) assert result.read() == b"file-content" @pytest.mark.asyncio async def test_cloudflare_read_leaves_raw_data_prefix_payload_unchanged() -> None: raw_payload = b'data: this is a normal file, not an SSE payload\n{"ok": false}\n' sess = _make_session( fake_http=_FakeHttp({"GET /file/": _FakeResponse(status=200, raw_body=raw_payload)}) ) result = await sess.read(Path("/workspace/test.txt")) assert result.read() == raw_payload @pytest.mark.asyncio async def test_cloudflare_read_rejects_truncated_streamed_file_payload() -> None: sess = _make_session( fake_http=_FakeHttp( { "GET /file/": _truncated_streamed_payload_response( payload=b"file-content", is_binary=False, ) } ) ) with pytest.raises(WorkspaceArchiveReadError): await sess.read(Path("/workspace/test.txt")) @pytest.mark.asyncio async def test_cloudflare_read_404_and_write_non_bytes_raise_structured_errors() -> None: fake_http = _FakeHttp( {"GET /file/": _FakeResponse(status=404, json_body={"error": "not found"})} ) sess = _make_session(fake_http=fake_http) with pytest.raises(WorkspaceReadNotFoundError): await sess.read(Path("/workspace/missing.txt")) class _BadIO(io.IOBase): def read(self, *args: Any) -> int: _ = args return 42 with pytest.raises(WorkspaceWriteTypeError): await sess.write(Path("/workspace/out.txt"), _BadIO()) @pytest.mark.asyncio async def test_cloudflare_read_and_write_normalize_workspace_paths() -> None: fake_http = _FakeHttp() sess = _make_session(fake_http=fake_http) with pytest.raises(InvalidManifestPathError): await sess.read(Path("../secret.txt")) with pytest.raises(InvalidManifestPathError): await sess.write(Path("/workspace/../secret.txt"), io.BytesIO(b"data")) assert fake_http.calls == [] @pytest.mark.asyncio async def test_cloudflare_persist_and_hydrate_use_http_endpoints() -> None: fake_http = _FakeHttp( { "POST /persist": _FakeResponse(status=200, raw_body=b"fake-tar"), "POST /hydrate": _FakeResponse(status=200, json_body={"ok": True}), } ) manifest = Manifest(entries={Path("cache"): Dir(ephemeral=True)}) sess = _make_session(state=_make_state(manifest=manifest), fake_http=fake_http) sess.register_persist_workspace_skip_path("generated/runtime") persisted = await sess.persist_workspace() assert persisted.read() == b"fake-tar" await sess.hydrate_workspace(io.BytesIO(_valid_tar_bytes())) persist_calls = [c for c in fake_http.calls if c["method"] == "POST" and "/persist" in c["url"]] hydrate_calls = [c for c in fake_http.calls if c["method"] == "POST" and "/hydrate" in c["url"]] assert "root" not in persist_calls[0]["params"] assert "cache" in persist_calls[0]["params"]["excludes"] assert "generated/runtime" in persist_calls[0]["params"]["excludes"] assert "root" not in hydrate_calls[0].get("params", {}) @pytest.mark.asyncio async def test_cloudflare_persist_retries_only_documented_503_status() -> None: fake_http = _FakeHttp( { "POST /persist": _FakeResponse( status=503, json_body={"error": "container starting"}, ) } ) sess = _make_session(fake_http=fake_http) with pytest.raises(WorkspaceArchiveReadError) as exc_info: await sess.persist_workspace() persist_calls = [c for c in fake_http.calls if c["method"] == "POST" and "/persist" in c["url"]] assert len(persist_calls) == 3 assert exc_info.value.context["http_status"] == 503 assert exc_info.value.retryable is True @pytest.mark.asyncio async def test_cloudflare_persist_does_not_retry_documented_fail_fast_500() -> None: fake_http = _FakeHttp( { "POST /persist": _FakeResponse( status=500, json_body={"error": "configuration error"}, ) } ) sess = _make_session(fake_http=fake_http) with pytest.raises(WorkspaceArchiveReadError) as exc_info: await sess.persist_workspace() persist_calls = [c for c in fake_http.calls if c["method"] == "POST" and "/persist" in c["url"]] assert len(persist_calls) == 1 assert exc_info.value.context["http_status"] == 500 assert exc_info.value.retryable is False @pytest.mark.asyncio async def test_cloudflare_persist_unmounts_and_remounts_ephemeral_bucket_mounts() -> None: fake_http = _FakeHttp( { "POST /mount": _FakeResponse(status=200, json_body={"ok": True}), "POST /unmount": _FakeResponse(status=200, json_body={"ok": True}), "POST /persist": _FakeResponse(status=200, raw_body=b"fake-tar"), } ) manifest = Manifest( entries={ "data": S3Mount( bucket="bucket", prefix="nested/prefix/", mount_strategy=CloudflareBucketMountStrategy(), ) } ) sess = _make_session(state=_make_state(manifest=manifest), fake_http=fake_http) persisted = await sess.persist_workspace() assert persisted.read() == b"fake-tar" assert [call["url"].split("/")[-1] for call in fake_http.calls] == [ "unmount", "persist", "mount", ] @pytest.mark.asyncio async def test_cloudflare_hydrate_unmounts_and_remounts_ephemeral_bucket_mounts() -> None: fake_http = _FakeHttp( { "POST /mount": _FakeResponse(status=200, json_body={"ok": True}), "POST /unmount": _FakeResponse(status=200, json_body={"ok": True}), "POST /hydrate": _FakeResponse(status=200, json_body={"ok": True}), } ) manifest = Manifest( entries={ "data": S3Mount( bucket="bucket", prefix="nested/prefix/", mount_strategy=CloudflareBucketMountStrategy(), ) } ) sess = _make_session(state=_make_state(manifest=manifest), fake_http=fake_http) await sess.hydrate_workspace(io.BytesIO(_valid_tar_bytes())) assert [call["url"].split("/")[-1] for call in fake_http.calls] == [ "unmount", "hydrate", "mount", ] @pytest.mark.asyncio async def test_cloudflare_resume_start_hydrates_without_preemptive_unmount() -> None: fake_http = _FakeHttp({"POST /hydrate": _FakeResponse(status=200, json_body={"ok": True})}) manifest = Manifest( entries={ "data": S3Mount( bucket="bucket", prefix="nested/prefix/", mount_strategy=CloudflareBucketMountStrategy(), ) } ) sess = _make_session(state=_make_state(manifest=manifest), fake_http=fake_http) sess.state.snapshot = _RestorableSnapshot(id="snapshot") sess.state.workspace_root_ready = True sess._start_workspace_root_ready = True # noqa: SLF001 sess._set_start_state_preserved(True) # noqa: SLF001 async def _exec_internal(*command: str | Path, timeout: float | None = None) -> ExecResult: _ = (command, timeout) return ExecResult(stdout=b"", stderr=b"", exit_code=0) sess._exec_internal = _exec_internal # type: ignore[method-assign] await sess.start() assert [call["url"].split("/")[-1] for call in fake_http.calls] == [ "running", "hydrate", "mount", ] @pytest.mark.asyncio async def test_cloudflare_resume_start_skips_hydrate_when_shared_resume_gate_matches() -> None: fake_http = _FakeHttp({"GET /running": _FakeResponse(status=200, json_body={"running": True})}) manifest = Manifest( entries={ "data": S3Mount( bucket="bucket", prefix="nested/prefix/", mount_strategy=CloudflareBucketMountStrategy(), ) } ) sess = _make_session(state=_make_state(manifest=manifest), fake_http=fake_http) sess.state.snapshot = _RestorableSnapshot(id="snapshot") sess.state.workspace_root_ready = True sess._start_workspace_root_ready = True # noqa: SLF001 sess._set_start_state_preserved(True) # noqa: SLF001 async def _exec_internal(*command: str | Path, timeout: float | None = None) -> ExecResult: _ = (command, timeout) return ExecResult(stdout=b"", stderr=b"", exit_code=0) async def _gate(*, is_running: bool) -> bool: assert is_running is True return True sess._exec_internal = _exec_internal # type: ignore[method-assign] sess._can_skip_snapshot_restore_on_resume = _gate # type: ignore[method-assign] await sess.start() assert [call["url"].split("/")[-1] for call in fake_http.calls] == [ "running", "mount", ] @pytest.mark.asyncio async def test_cloudflare_resume_start_unmounts_before_hydrate_when_sandbox_is_running() -> None: fake_http = _FakeHttp( { "GET /running": _FakeResponse(status=200, json_body={"running": True}), "POST /unmount": _FakeResponse(status=200, json_body={"ok": True}), "POST /hydrate": _FakeResponse(status=200, json_body={"ok": True}), } ) manifest = Manifest( entries={ "data": S3Mount( bucket="bucket", prefix="nested/prefix/", mount_strategy=CloudflareBucketMountStrategy(), ) } ) sess = _make_session(state=_make_state(manifest=manifest), fake_http=fake_http) sess.state.snapshot = _RestorableSnapshot(id="snapshot") sess.state.workspace_root_ready = True sess._start_workspace_root_ready = True # noqa: SLF001 sess._set_start_state_preserved(True) # noqa: SLF001 async def _exec_internal(*command: str | Path, timeout: float | None = None) -> ExecResult: _ = (command, timeout) return ExecResult(stdout=b"", stderr=b"", exit_code=0) sess._exec_internal = _exec_internal # type: ignore[method-assign] await sess.start() assert [call["url"].split("/")[-1] for call in fake_http.calls] == [ "running", "unmount", "hydrate", "mount", ] @pytest.mark.asyncio async def test_cloudflare_persist_preserves_hidden_exclude_paths() -> None: fake_http = _FakeHttp({"POST /persist": _FakeResponse(status=200, raw_body=b"fake-tar")}) sess = _make_session(fake_http=fake_http) sess.register_persist_workspace_skip_path(".sandbox-blobfuse-config/session") sess.register_persist_workspace_skip_path("./generated/runtime") await sess.persist_workspace() persist_calls = [c for c in fake_http.calls if c["method"] == "POST" and "/persist" in c["url"]] assert persist_calls[0]["params"]["excludes"].split(",") == [ ".sandbox-blobfuse-config/session", "generated/runtime", ] @pytest.mark.asyncio async def test_cloudflare_persist_decodes_streamed_archive_payload() -> None: fake_http = _FakeHttp( {"POST /persist": _streamed_payload_response(payload=b"fake-tar", is_binary=True)} ) sess = _make_session(fake_http=fake_http) persisted = await sess.persist_workspace() assert persisted.read() == b"fake-tar" @pytest.mark.asyncio async def test_cloudflare_persist_leaves_raw_data_prefix_archive_unchanged() -> None: raw_payload = b"data: raw tar bytes that happen to share the prefix" fake_http = _FakeHttp({"POST /persist": _FakeResponse(status=200, raw_body=raw_payload)}) sess = _make_session(fake_http=fake_http) persisted = await sess.persist_workspace() assert persisted.read() == raw_payload @pytest.mark.asyncio async def test_cloudflare_persist_rejects_truncated_streamed_archive_payload() -> None: fake_http = _FakeHttp( {"POST /persist": _truncated_streamed_payload_response(payload=b"fake-tar", is_binary=True)} ) sess = _make_session(fake_http=fake_http) with pytest.raises(WorkspaceArchiveReadError): await sess.persist_workspace() @pytest.mark.asyncio async def test_cloudflare_delete_calls_shutdown() -> None: fake_http = _FakeHttp() inner = _make_session(state=_make_state(), fake_http=fake_http) client = CloudflareSandboxClient() session = client._wrap_session(inner) await client.delete(session) delete_calls = [c for c in fake_http.calls if c["method"] == "DELETE"] assert len(delete_calls) == 1 @pytest.mark.asyncio async def test_cloudflare_supports_pty() -> None: sess = _make_session() assert sess.supports_pty() is True @pytest.mark.asyncio async def test_cloudflare_pty_exec_start_opens_websocket_and_sends_command() -> None: fake_http = _FakeHttp() fake_http.fake_ws = _FakeWebSocket( frames=[ _ws_text_frame({"type": "ready"}), _ws_binary_frame(b">>> "), _ws_text_frame({"type": "exit", "code": 0}), ] ) sess = _make_session(fake_http=fake_http) started = await sess.pty_exec_start("python3", shell=False, tty=True, yield_time_s=0.05) assert started.process_id is None assert started.exit_code == 0 assert started.output == b">>> " assert fake_http.ws_connect_calls == [ {"url": "wss://sandbox-cf.example.workers.dev/v1/sandbox/abc123/pty?cols=80&rows=24"} ] assert fake_http.fake_ws.sent_bytes == [b"python3\n"] assert fake_http.fake_ws.closed is True @pytest.mark.asyncio async def test_cloudflare_pty_write_stdin_sends_input_and_collects_output() -> None: fake_ws = _FakeWebSocket() sess = _make_session(fake_http=_FakeHttp()) process_id = await _register_pty_entry(sess, ws=fake_ws, tty=True) entry = sess._pty_processes[process_id] async with entry.output_lock: entry.output_chunks.append(b"10\n") entry.output_notify.set() updated = await sess.pty_write_stdin( session_id=process_id, chars="5 + 5\n", yield_time_s=0.05, ) assert updated.process_id == process_id assert updated.exit_code is None assert updated.output == b"10\n" assert fake_ws.sent_bytes == [b"5 + 5\n"] @pytest.mark.asyncio async def test_cloudflare_pty_write_stdin_rejects_unknown_session() -> None: sess = _make_session(fake_http=_FakeHttp()) with pytest.raises(PtySessionNotFoundError): await sess.pty_write_stdin(session_id=999_999, chars="") @pytest.mark.asyncio async def test_cloudflare_pty_write_stdin_rejects_non_tty_input() -> None: fake_ws = _FakeWebSocket() sess = _make_session(fake_http=_FakeHttp()) process_id = await _register_pty_entry(sess, ws=fake_ws, tty=False) with pytest.raises(RuntimeError, match="stdin is not available for this process"): await sess.pty_write_stdin(session_id=process_id, chars="hello") @pytest.mark.asyncio async def test_cloudflare_pty_terminate_all_closes_websockets() -> None: sess = _make_session(fake_http=_FakeHttp()) fake_ws_1 = _FakeWebSocket() fake_ws_2 = _FakeWebSocket() await _register_pty_entry(sess, ws=fake_ws_1, tty=True) await _register_pty_entry(sess, ws=fake_ws_2, tty=True) await sess.pty_terminate_all() assert sess._pty_processes == {} assert sess._reserved_pty_process_ids == set() assert fake_ws_1.closed is True assert fake_ws_2.closed is True @pytest.mark.asyncio async def test_cloudflare_pty_exec_start_prunes_oldest_session() -> None: fake_http = _FakeHttp() sess = _make_session(fake_http=fake_http) oldest_ws = _FakeWebSocket() await _register_pty_entry(sess, ws=oldest_ws, tty=True, last_used=0.0) for index in range(1, PTY_PROCESSES_MAX): await _register_pty_entry( sess, ws=_FakeWebSocket(), tty=True, last_used=float(index), ) fake_http.fake_ws = _BlockingFakeWebSocket(frames=[_ws_text_frame({"type": "ready"})]) started = await sess.pty_exec_start("python3", shell=False, tty=True, yield_time_s=0.05) assert started.process_id is not None assert oldest_ws.closed is True assert len(sess._pty_processes) == PTY_PROCESSES_MAX @pytest.mark.asyncio async def test_cloudflare_pty_exec_start_wraps_websocket_connect_failures() -> None: class _FailingHttp(_FakeHttp): async def ws_connect(self, url: str, **kwargs: Any) -> _FakeWebSocket: _ = (url, kwargs) raise aiohttp.ClientError("connect failed") sess = _make_session(fake_http=_FailingHttp()) with pytest.raises(ExecTransportError) as exc_info: await sess.pty_exec_start("python3", shell=False, tty=True) assert isinstance(exc_info.value.__cause__, aiohttp.ClientError) assert str(exc_info.value.__cause__) == "connect failed" assert str(exc_info.value) == ( "Cloudflare pty exec transport failed: ClientError: connect failed" ) assert exc_info.value.context["backend"] == "cloudflare" assert exc_info.value.context["operation"] == "pty exec" assert exc_info.value.context["provider_error"] == "ClientError: connect failed" @pytest.mark.asyncio async def test_cloudflare_pty_exec_start_wraps_ready_timeout() -> None: class _NeverReadyWebSocket(_FakeWebSocket): async def receive(self) -> aiohttp.WSMessage: raise asyncio.TimeoutError() fake_http = _FakeHttp() fake_http.fake_ws = _NeverReadyWebSocket() sess = _make_session(fake_http=fake_http) with pytest.raises(ExecTimeoutError): await sess.pty_exec_start("python3", shell=False, tty=True) assert fake_http.fake_ws.closed is True @pytest.mark.asyncio async def test_cloudflare_stop_terminates_active_pty_sessions() -> None: fake_http = _FakeHttp({"POST /persist": _FakeResponse(status=200, raw_body=b"fake-tar")}) sess = _make_session(fake_http=fake_http) fake_ws = _FakeWebSocket() process_id = await _register_pty_entry(sess, ws=fake_ws, tty=True) await sess.stop() assert fake_ws.closed is True with pytest.raises(PtySessionNotFoundError): await sess.pty_write_stdin(session_id=process_id, chars="") @pytest.mark.asyncio async def test_cloudflare_hydrate_rejects_unsafe_tar() -> None: """Verify that _hydrate_workspace_via_http rejects archives with path-traversal members.""" buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w") as tar: info = tarfile.TarInfo(name="../../etc/passwd") info.size = 5 tar.addfile(info, io.BytesIO(b"evil\n")) buf.seek(0) fake_http = _FakeHttp({"POST /hydrate": _FakeResponse(status=200, json_body={"ok": True})}) sess = _make_session(fake_http=fake_http) from agents.sandbox.errors import WorkspaceArchiveWriteError with pytest.raises(WorkspaceArchiveWriteError) as exc_info: await sess._hydrate_workspace_via_http(buf) assert exc_info.value.context.get("reason") == "unsafe_or_invalid_tar" assert exc_info.value.context.get("member") is not None # The HTTP POST should never have been made. assert not any(c["method"] == "POST" and "/hydrate" in c["url"] for c in fake_http.calls) def test_cloudflare_runtime_helpers_returns_resolve_helper() -> None: """Verify that _runtime_helpers() includes the workspace path resolver.""" from agents.sandbox.session.runtime_helpers import RESOLVE_WORKSPACE_PATH_HELPER sess = _make_session() helpers = sess._runtime_helpers() assert RESOLVE_WORKSPACE_PATH_HELPER in helpers assert sess._current_runtime_helper_cache_key() == sess.state.sandbox_id @pytest.mark.asyncio async def test_cloudflare_read_validates_path_access() -> None: """Verify that read() routes through _validate_path_access for symlink safety.""" fake_http = _FakeHttp({"GET /file/": _FakeResponse(status=200, raw_body=b"file-content")}) sess = _make_session(fake_http=fake_http) calls: list[tuple[str, bool]] = [] async def _tracking_normalize(path: Path | str, *, for_write: bool = False) -> Path: calls.append((Path(path).as_posix(), for_write)) # Fall back to synchronous normalize_path to avoid needing a real remote. return sess.normalize_path(path, for_write=for_write) sess._validate_path_access = _tracking_normalize # type: ignore[method-assign] await sess.read(Path("/workspace/test.txt")) assert calls == [("/workspace/test.txt", False)] @pytest.mark.asyncio async def test_cloudflare_write_validates_path_access_for_write() -> None: """Verify that write() routes through _validate_path_access(for_write=True).""" fake_http = _FakeHttp({"PUT /file/": _FakeResponse(status=200, json_body={"ok": True})}) sess = _make_session(fake_http=fake_http) calls: list[tuple[str, bool]] = [] async def _tracking_normalize(path: Path | str, *, for_write: bool = False) -> Path: calls.append((Path(path).as_posix(), for_write)) return sess.normalize_path(path, for_write=for_write) sess._validate_path_access = _tracking_normalize # type: ignore[method-assign] await sess.write(Path("/workspace/out.txt"), io.BytesIO(b"data")) assert calls == [("/workspace/out.txt", True)] @pytest.mark.asyncio async def test_cloudflare_shutdown_logs_on_failure(caplog: pytest.LogCaptureFixture) -> None: """Verify that _shutdown_backend logs at DEBUG when the DELETE request fails.""" import logging class _FailingDeleteHttp(_FakeHttp): def delete(self, url: str, **kwargs: Any) -> Any: raise aiohttp.ClientError("delete failed") sess = _make_session(fake_http=_FailingDeleteHttp()) with caplog.at_level(logging.DEBUG, logger="agents.extensions.sandbox.cloudflare.sandbox"): await sess._shutdown_backend() assert any("Failed to delete Cloudflare sandbox" in r.message for r in caplog.records) @pytest.mark.asyncio async def test_cloudflare_shutdown_logs_delete_response_details( caplog: pytest.LogCaptureFixture, ) -> None: """Verify that DELETE response bodies are kept when shutdown cleanup fails.""" import logging sess = _make_session( fake_http=_FakeHttp( { "DELETE /v1/sandbox/": _FakeResponse( status=502, json_body={ "error": "pool error: Failed to start container", "code": "pool_error", }, ) } ) ) with caplog.at_level(logging.DEBUG, logger="agents.extensions.sandbox.cloudflare.sandbox"): await sess._shutdown_backend() assert any( "DELETE /sandbox failed: HTTP 502: pool_error: pool error: Failed to start container" in r.message for r in caplog.records )