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
    )
