"""
Daytona sandbox (https://daytona.io) implementation.

This module provides a Daytona-backed sandbox client/session implementation backed by
`daytona.Sandbox` via the AsyncDaytona client.

The `daytona` dependency is optional, so package-level exports should guard imports of this
module. Within this module, Daytona SDK imports are lazy so users without the extra can still
import the package.
"""

from __future__ import annotations

import asyncio
import io
import logging
import math
import shlex
import time
import uuid
from collections import deque
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Literal, cast
from urllib.parse import urlsplit

from pydantic import BaseModel, Field

from ....sandbox.entries import Mount
from ....sandbox.errors import (
    ExecTimeoutError,
    ExecTransportError,
    ExposedPortUnavailableError,
    InvalidManifestPathError as InvalidManifestPathError,
    WorkspaceArchiveReadError,
    WorkspaceArchiveWriteError,
    WorkspaceReadNotFoundError,
    WorkspaceStartError,
    WorkspaceWriteTypeError,
)
from ....sandbox.manifest import Manifest
from ....sandbox.session import SandboxSession, SandboxSessionState
from ....sandbox.session.base_sandbox_session import BaseSandboxSession
from ....sandbox.session.dependencies import Dependencies
from ....sandbox.session.manager import Instrumentation
from ....sandbox.session.pty_types import (
    PTY_PROCESSES_MAX,
    PTY_PROCESSES_WARNING,
    PtyExecUpdate,
    allocate_pty_process_id,
    clamp_pty_yield_time_ms,
    process_id_to_prune_from_meta,
    resolve_pty_write_yield_time_ms,
    truncate_text_by_tokens,
)
from ....sandbox.session.runtime_helpers import RESOLVE_WORKSPACE_PATH_HELPER, RuntimeHelperScript
from ....sandbox.session.sandbox_client import BaseSandboxClient, BaseSandboxClientOptions
from ....sandbox.session.tar_workspace import shell_tar_exclude_args
from ....sandbox.snapshot import SnapshotBase, SnapshotSpec, resolve_snapshot
from ....sandbox.types import ExecResult, ExposedPortEndpoint, User
from ....sandbox.util.retry import (
    TRANSIENT_HTTP_STATUS_CODES,
    exception_chain_contains_type,
    exception_chain_has_status_code,
    retry_async,
)
from ....sandbox.util.tar_utils import UnsafeTarMemberError, validate_tar_bytes
from ....sandbox.workspace_paths import (
    coerce_posix_path,
    posix_path_as_path,
    posix_path_for_error,
    sandbox_path_str,
)

DEFAULT_DAYTONA_WORKSPACE_ROOT = "/home/daytona/workspace"
logger = logging.getLogger(__name__)


def _daytona_provider_error_detail(error: BaseException) -> str | None:
    message = str(error)
    status = getattr(error, "status_code", None) or getattr(error, "status", None)
    if isinstance(status, int):
        if message:
            return f"HTTP {status}: {message}"
        return f"HTTP {status}"
    if message:
        return f"{type(error).__name__}: {message}"
    return type(error).__name__


def _daytona_exec_transport_error(
    *,
    command: tuple[str | Path, ...],
    cause: BaseException,
) -> ExecTransportError:
    detail = _daytona_provider_error_detail(cause)
    context: dict[str, object] = {"backend": "daytona"}
    if detail:
        context["provider_error"] = detail
    status = getattr(cause, "status_code", None) or getattr(cause, "status", None)
    if isinstance(status, int):
        context["http_status"] = status
    message = "Daytona exec failed"
    if detail:
        message = f"{message}: {detail}"
    return ExecTransportError(command=command, context=context, cause=cause, message=message)


def _import_daytona_sdk() -> tuple[Any, Any, Any, Any]:
    """Lazily import Daytona SDK classes, raising a clear error if missing."""
    try:
        from daytona import (
            AsyncDaytona,
            CreateSandboxFromImageParams,
            CreateSandboxFromSnapshotParams,
            DaytonaConfig,
        )

        return (
            AsyncDaytona,
            DaytonaConfig,
            CreateSandboxFromSnapshotParams,
            CreateSandboxFromImageParams,
        )
    except ImportError as e:
        raise ImportError(
            "DaytonaSandboxClient requires the optional `daytona` dependency.\n"
            "Install the Daytona extra before using this sandbox backend."
        ) from e


def _import_sandbox_state() -> Any:
    """Lazily import SandboxState enum from Daytona SDK, or None if unavailable."""
    try:
        from daytona import SandboxState

        return SandboxState
    except ImportError:
        return None


def _import_sdk_resources() -> Any:
    """Lazily import Resources from Daytona SDK."""
    try:
        from daytona import Resources

        return Resources
    except ImportError as e:
        raise ImportError(
            "DaytonaSandboxClient requires the optional `daytona` dependency.\n"
            "Install the Daytona extra before using this sandbox backend."
        ) from e


def _import_pty_size() -> Any:
    """Lazily import PtySize from Daytona SDK."""
    try:
        from daytona.common.pty import PtySize

        return PtySize
    except ImportError as e:
        raise ImportError(
            "DaytonaSandboxClient requires the optional `daytona` dependency.\n"
            "Install the Daytona extra before using this sandbox backend."
        ) from e


def _import_session_execute_request() -> Any:
    """Lazily import SessionExecuteRequest from Daytona SDK."""
    try:
        from daytona import SessionExecuteRequest

        return SessionExecuteRequest
    except ImportError as e:
        raise ImportError(
            "DaytonaSandboxClient requires the optional `daytona` dependency.\n"
            "Install the Daytona extra before using this sandbox backend."
        ) from e


def _import_daytona_exceptions() -> dict[str, type[BaseException]]:
    """Best-effort import Daytona exception classes for fine-grained error mapping."""
    try:
        from daytona import (
            DaytonaError,
            DaytonaNotFoundError,
            DaytonaRateLimitError,
            DaytonaTimeoutError,
        )
    except Exception:
        return {}
    return {
        "base": DaytonaError,
        "timeout": DaytonaTimeoutError,
        "not_found": DaytonaNotFoundError,
        "rate_limit": DaytonaRateLimitError,
    }


def _retryable_persist_workspace_error_types() -> tuple[type[BaseException], ...]:
    excs = _import_daytona_exceptions()
    retryable: list[type[BaseException]] = [asyncio.TimeoutError]
    timeout_exc = excs.get("timeout")
    if timeout_exc is not None:
        retryable.append(timeout_exc)
    return tuple(retryable)


class DaytonaSandboxResources(BaseModel):
    """Resource configuration for a Daytona sandbox."""

    model_config = {"frozen": True}

    cpu: int | None = None
    memory: int | None = None
    disk: int | None = None


class DaytonaSandboxTimeouts(BaseModel):
    """Timeout configuration for Daytona sandbox operations."""

    exec_timeout_unbounded_s: int = Field(default=24 * 60 * 60, ge=1)
    keepalive_s: int = Field(default=10, ge=1)
    cleanup_s: int = Field(default=30, ge=1)
    fast_op_s: int = Field(default=30, ge=1)
    file_upload_s: int = Field(default=1800, ge=1)
    file_download_s: int = Field(default=1800, ge=1)
    workspace_tar_s: int = Field(default=300, ge=1)


class DaytonaSandboxClientOptions(BaseSandboxClientOptions):
    """Client options for the Daytona sandbox."""

    type: Literal["daytona"] = "daytona"
    sandbox_snapshot_name: str | None = None
    image: str | None = None
    resources: DaytonaSandboxResources | None = None
    env_vars: dict[str, str] | None = None
    pause_on_exit: bool = False
    create_timeout: int = 60
    start_timeout: int = 60
    name: str | None = None
    auto_stop_interval: int = 0
    timeouts: DaytonaSandboxTimeouts | dict[str, object] | None = None
    exposed_ports: tuple[int, ...] = ()
    # This TTL applies to new connection setup only: Daytona checks signed preview URL expiry during
    # the initial HTTP request / websocket upgrade handshake. In live testing, an already-open
    # websocket stayed connected after the URL expired, but any reconnect or new handshake needed a
    # freshly resolved URL.
    exposed_port_url_ttl_s: int = 3600

    def __init__(
        self,
        sandbox_snapshot_name: str | None = None,
        image: str | None = None,
        resources: DaytonaSandboxResources | None = None,
        env_vars: dict[str, str] | None = None,
        pause_on_exit: bool = False,
        create_timeout: int = 60,
        start_timeout: int = 60,
        name: str | None = None,
        auto_stop_interval: int = 0,
        timeouts: DaytonaSandboxTimeouts | dict[str, object] | None = None,
        exposed_ports: tuple[int, ...] = (),
        exposed_port_url_ttl_s: int = 3600,
        *,
        type: Literal["daytona"] = "daytona",
    ) -> None:
        super().__init__(
            type=type,
            sandbox_snapshot_name=sandbox_snapshot_name,
            image=image,
            resources=resources,
            env_vars=env_vars,
            pause_on_exit=pause_on_exit,
            create_timeout=create_timeout,
            start_timeout=start_timeout,
            name=name,
            auto_stop_interval=auto_stop_interval,
            timeouts=timeouts,
            exposed_ports=exposed_ports,
            exposed_port_url_ttl_s=exposed_port_url_ttl_s,
        )


class DaytonaSandboxSessionState(SandboxSessionState):
    """Serializable state for a Daytona-backed session."""

    type: Literal["daytona"] = "daytona"
    sandbox_id: str
    sandbox_snapshot_name: str | None = None
    image: str | None = None
    base_env_vars: dict[str, str] = Field(default_factory=dict)
    pause_on_exit: bool = False
    create_timeout: int = 60
    start_timeout: int = 60
    name: str | None = None
    resources: DaytonaSandboxResources | None = None
    auto_stop_interval: int = 0
    timeouts: DaytonaSandboxTimeouts = Field(default_factory=DaytonaSandboxTimeouts)
    exposed_port_url_ttl_s: int = 3600


@dataclass
class _DaytonaPtySessionEntry:
    daytona_session_id: str
    pty_handle: Any
    tty: bool = True
    cmd_id: str | None = None
    output_chunks: deque[bytes] = field(default_factory=deque)
    output_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
    output_notify: asyncio.Event = field(default_factory=asyncio.Event)
    last_used: float = field(default_factory=time.monotonic)
    done: bool = False
    exit_code: int | None = None


class DaytonaSandboxSession(BaseSandboxSession):
    """Daytona-backed sandbox session implementation."""

    state: DaytonaSandboxSessionState
    _sandbox: Any
    _pty_lock: asyncio.Lock
    _pty_sessions: dict[int, _DaytonaPtySessionEntry]
    _reserved_pty_process_ids: set[int]

    def __init__(self, *, state: DaytonaSandboxSessionState, sandbox: Any) -> None:
        self.state = state
        self._sandbox = sandbox
        self._pty_lock = asyncio.Lock()
        self._pty_sessions = {}
        self._reserved_pty_process_ids = set()

    @classmethod
    def from_state(
        cls,
        state: DaytonaSandboxSessionState,
        *,
        sandbox: Any,
    ) -> DaytonaSandboxSession:
        return cls(state=state, sandbox=sandbox)

    @property
    def sandbox_id(self) -> str:
        return self.state.sandbox_id

    async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
        try:
            preview = await self._sandbox.create_signed_preview_url(
                port,
                expires_in_seconds=self.state.exposed_port_url_ttl_s,
            )
        except Exception as e:
            raise ExposedPortUnavailableError(
                port=port,
                exposed_ports=self.state.exposed_ports,
                reason="backend_unavailable",
                context={"backend": "daytona", "detail": "create_signed_preview_url_failed"},
                cause=e,
            ) from e

        url = getattr(preview, "url", None)
        if not isinstance(url, str) or not url:
            raise ExposedPortUnavailableError(
                port=port,
                exposed_ports=self.state.exposed_ports,
                reason="backend_unavailable",
                context={"backend": "daytona", "detail": "invalid_preview_url", "url": url},
            )

        try:
            split = urlsplit(url)
            host = split.hostname
            if host is None:
                raise ValueError("missing hostname")
            port_value = split.port or (443 if split.scheme == "https" else 80)
            return ExposedPortEndpoint(host=host, port=port_value, tls=split.scheme == "https")
        except Exception as e:
            raise ExposedPortUnavailableError(
                port=port,
                exposed_ports=self.state.exposed_ports,
                reason="backend_unavailable",
                context={"backend": "daytona", "detail": "invalid_preview_url", "url": url},
                cause=e,
            ) from e

    async def _shutdown_backend(self) -> None:
        try:
            if self.state.pause_on_exit:
                await self._sandbox.stop()
            else:
                await self._sandbox.delete()
        except Exception:
            pass

    async def _validate_path_access(self, path: Path | str, *, for_write: bool = False) -> Path:
        return await self._validate_remote_path_access(path, for_write=for_write)

    def _runtime_helpers(self) -> tuple[RuntimeHelperScript, ...]:
        return (RESOLVE_WORKSPACE_PATH_HELPER,)

    async def _prepare_workspace_root(self) -> None:
        """Create the workspace root before SDK exec calls use it as cwd."""
        root = sandbox_path_str(self.state.manifest.root)
        error_root = posix_path_for_error(root)
        try:
            envs = await self._resolved_envs()
            result = await self._sandbox.process.exec(
                f"mkdir -p -- {shlex.quote(root)}",
                env=envs or None,
                timeout=self.state.timeouts.fast_op_s,
            )
        except Exception as e:
            detail = _daytona_provider_error_detail(e)
            message = "failed to start session"
            if detail:
                message = f"{message}: Daytona workspace root setup failed: {detail}"
            raise WorkspaceStartError(
                path=error_root,
                context={"backend": "daytona", "reason": "workspace_root_setup_failed"},
                cause=e,
                message=message,
            ) from e

        exit_code = int(getattr(result, "exit_code", 0) or 0)
        if exit_code != 0:
            output = str(getattr(result, "result", "") or "")
            message = (
                f"failed to start session: Daytona workspace root setup exited with {exit_code}"
            )
            if output:
                message = f"{message}: {output}"
            raise WorkspaceStartError(
                path=error_root,
                context={
                    "backend": "daytona",
                    "reason": "workspace_root_nonzero_exit",
                    "exit_code": exit_code,
                    "output": output,
                },
                message=message,
            )

    async def _prepare_backend_workspace(self) -> None:
        await self._prepare_workspace_root()

    async def mkdir(
        self,
        path: Path | str,
        *,
        parents: bool = False,
        user: str | User | None = None,
    ) -> None:
        if user is not None:
            path = await self._check_mkdir_with_exec(path, parents=parents, user=user)
        else:
            path = await self._validate_path_access(path, for_write=True)
        if path == Path("/"):
            return
        try:
            await self._sandbox.fs.create_folder(sandbox_path_str(path), "755")
        except Exception as e:
            raise WorkspaceArchiveWriteError(
                path=path,
                context={"reason": "mkdir_failed"},
                cause=e,
            ) from e

    async def _resolved_envs(self) -> dict[str, str]:
        manifest_envs = await self.state.manifest.environment.resolve()
        return {**self.state.base_env_vars, **manifest_envs}

    def _coerce_exec_timeout(self, timeout_s: float | None) -> float:
        if timeout_s is None:
            return float(self.state.timeouts.exec_timeout_unbounded_s)
        if timeout_s <= 0:
            return 0.001
        return float(timeout_s)

    async def _exec_internal(
        self,
        *command: str | Path,
        timeout: float | None = None,
    ) -> ExecResult:
        cmd_str = shlex.join(str(c) for c in command)
        envs = await self._resolved_envs()
        cwd = sandbox_path_str(self.state.manifest.root)
        env_args = (
            " ".join(shlex.quote(f"{key}={value}") for key, value in envs.items()) if envs else ""
        )
        env_wrapper = f"env -- {env_args} " if env_args else ""
        session_cmd = f"cd {shlex.quote(cwd)} && {env_wrapper}{cmd_str}"
        daytona_session_id = f"sandbox-{uuid.uuid4().hex[:12]}"

        caller_timeout = self._coerce_exec_timeout(timeout)
        deadline = time.monotonic() + caller_timeout
        SessionExecuteRequest = _import_session_execute_request()
        daytona_exc = _import_daytona_exceptions()
        timeout_exc = daytona_exc.get("timeout")

        def _remaining_timeout() -> float:
            return max(0.0, deadline - time.monotonic())

        try:
            await asyncio.wait_for(
                self._sandbox.process.create_session(daytona_session_id),
                timeout=_remaining_timeout(),
            )
            command_timeout = _remaining_timeout()
            sdk_timeout = max(1, math.ceil(command_timeout + 1.0))
            result = await asyncio.wait_for(
                self._sandbox.process.execute_session_command(
                    daytona_session_id,
                    SessionExecuteRequest(command=session_cmd, run_async=False),
                    timeout=sdk_timeout,
                ),
                timeout=caller_timeout,
            )
            exit_code = int(result.exit_code or 0)
            stdout = getattr(result, "stdout", None)
            stderr = getattr(result, "stderr", None)
            if stdout is None and stderr is None:
                output = getattr(result, "output", "") or ""
                if exit_code == 0:
                    stdout = output
                    stderr = ""
                else:
                    stdout = ""
                    stderr = output
            return ExecResult(
                stdout=(stdout or "").encode("utf-8", errors="replace"),
                stderr=(stderr or "").encode("utf-8", errors="replace"),
                exit_code=exit_code,
            )
        except asyncio.TimeoutError as e:
            raise ExecTimeoutError(command=command, timeout_s=timeout, cause=e) from e
        except Exception as e:
            if timeout_exc is not None and isinstance(e, timeout_exc):
                raise ExecTimeoutError(command=command, timeout_s=timeout, cause=e) from e
            raise _daytona_exec_transport_error(command=command, cause=e) from e
        finally:
            try:
                await asyncio.wait_for(
                    self._sandbox.process.delete_session(daytona_session_id),
                    timeout=self.state.timeouts.cleanup_s,
                )
            except Exception:
                pass

    def supports_pty(self) -> bool:
        return True

    async def pty_exec_start(
        self,
        *command: str | Path,
        timeout: float | None = None,
        shell: bool | list[str] = True,
        user: str | User | None = None,
        tty: bool = False,
        yield_time_s: float | None = None,
        max_output_tokens: int | None = None,
    ) -> PtyExecUpdate:
        PtySize = _import_pty_size()
        sanitized = self._prepare_exec_command(*command, shell=shell, user=user)
        cmd_str = shlex.join(str(part) for part in sanitized)
        envs = await self._resolved_envs()
        cwd = sandbox_path_str(self.state.manifest.root)
        exec_timeout = self._coerce_exec_timeout(timeout)
        daytona_exc = _import_daytona_exceptions()
        timeout_exc = daytona_exc.get("timeout")

        daytona_session_id = f"sandbox-{uuid.uuid4().hex[:12]}"
        entry = _DaytonaPtySessionEntry(
            daytona_session_id=daytona_session_id,
            pty_handle=None,
            tty=tty,
        )

        async def _on_data(chunk: bytes | str) -> None:
            raw = (
                chunk.encode("utf-8", errors="replace") if isinstance(chunk, str) else bytes(chunk)
            )
            async with entry.output_lock:
                entry.output_chunks.append(raw)
            entry.output_notify.set()

        pruned: _DaytonaPtySessionEntry | None = None
        registered = False
        try:
            if tty:
                pty_handle = await asyncio.wait_for(
                    self._sandbox.process.create_pty_session(
                        id=daytona_session_id,
                        on_data=_on_data,
                        cwd=cwd,
                        envs=envs or None,
                        pty_size=PtySize(cols=80, rows=24),
                    ),
                    timeout=exec_timeout,
                )
                entry.pty_handle = pty_handle
                asyncio.create_task(self._run_pty_waiter(entry))
                await asyncio.wait_for(pty_handle.wait_for_connection(), timeout=exec_timeout)
                await asyncio.wait_for(
                    pty_handle.send_input(cmd_str + "\n"),
                    timeout=self.state.timeouts.fast_op_s,
                )
            else:
                SessionExecuteRequest = _import_session_execute_request()
                env_args = (
                    " ".join(shlex.quote(f"{key}={value}") for key, value in envs.items())
                    if envs
                    else ""
                )
                env_wrapper = f"env -- {env_args} " if env_args else ""
                session_cmd = f"cd {shlex.quote(cwd)} && {env_wrapper}{cmd_str}"
                await asyncio.wait_for(
                    self._sandbox.process.create_session(daytona_session_id),
                    timeout=exec_timeout,
                )
                resp = await asyncio.wait_for(
                    self._sandbox.process.execute_session_command(
                        daytona_session_id,
                        SessionExecuteRequest(command=session_cmd, run_async=True),
                    ),
                    timeout=exec_timeout,
                )
                entry.cmd_id = resp.cmd_id
                asyncio.create_task(
                    self._run_session_reader(
                        entry,
                        daytona_session_id,
                        resp.cmd_id,
                        _on_data,
                    )
                )

            async with self._pty_lock:
                process_id = allocate_pty_process_id(self._reserved_pty_process_ids)
                self._reserved_pty_process_ids.add(process_id)
                pruned = self._prune_pty_sessions_if_needed()
                self._pty_sessions[process_id] = entry
                process_count = len(self._pty_sessions)
                registered = True
        except asyncio.TimeoutError as e:
            if not registered:
                cleanup_task = asyncio.ensure_future(self._terminate_pty_entry(entry))
                try:
                    await asyncio.shield(cleanup_task)
                except BaseException:
                    await asyncio.shield(cleanup_task)
            raise ExecTimeoutError(command=command, timeout_s=timeout, cause=e) from e
        except Exception as e:
            if not registered:
                cleanup_task = asyncio.ensure_future(self._terminate_pty_entry(entry))
                try:
                    await asyncio.shield(cleanup_task)
                except BaseException:
                    await asyncio.shield(cleanup_task)
            if timeout_exc is not None and isinstance(e, timeout_exc):
                raise ExecTimeoutError(command=command, timeout_s=timeout, cause=e) from e
            raise _daytona_exec_transport_error(command=command, cause=e) from e
        except BaseException:
            if not registered:
                cleanup_task = asyncio.ensure_future(self._terminate_pty_entry(entry))
                try:
                    await asyncio.shield(cleanup_task)
                except BaseException:
                    await asyncio.shield(cleanup_task)
            raise

        if pruned is not None:
            await self._terminate_pty_entry(pruned)

        if process_count >= PTY_PROCESSES_WARNING:
            logger.warning(
                "PTY process count reached warning threshold: %s active sessions",
                process_count,
            )

        yield_time_ms = 10_000 if yield_time_s is None else int(yield_time_s * 1000)
        output, original_token_count = await self._collect_pty_output(
            entry=entry,
            yield_time_ms=clamp_pty_yield_time_ms(yield_time_ms),
            max_output_tokens=max_output_tokens,
        )
        return await self._finalize_pty_update(
            process_id=process_id,
            entry=entry,
            output=output,
            original_token_count=original_token_count,
        )

    async def _run_pty_waiter(self, entry: _DaytonaPtySessionEntry) -> None:
        try:
            await entry.pty_handle.wait()
            ec = getattr(entry.pty_handle, "exit_code", None)
            if ec is not None:
                entry.exit_code = int(ec)
        except Exception:
            pass
        finally:
            entry.done = True
            entry.output_notify.set()

    async def _run_session_reader(
        self,
        entry: _DaytonaPtySessionEntry,
        session_id: str,
        cmd_id: str,
        on_data: Any,
    ) -> None:
        logs_failed = False
        try:
            await self._sandbox.process.get_session_command_logs_async(
                session_id,
                cmd_id,
                on_data,
                on_data,
            )
        except Exception:
            logs_failed = True
        finally:
            try:
                cmd = await self._sandbox.process.get_session_command(session_id, cmd_id)
                if cmd.exit_code is not None:
                    entry.exit_code = int(cmd.exit_code)
                    entry.done = True
            except Exception:
                pass
            if not logs_failed:
                entry.done = True
            entry.output_notify.set()

    async def pty_write_stdin(
        self,
        *,
        session_id: int,
        chars: str,
        yield_time_s: float | None = None,
        max_output_tokens: int | None = None,
    ) -> PtyExecUpdate:
        async with self._pty_lock:
            entry = self._resolve_pty_session_entry(
                pty_processes=self._pty_sessions,
                session_id=session_id,
            )

        if chars:
            if not entry.tty:
                raise RuntimeError("stdin is not available for this process")
            await asyncio.wait_for(
                entry.pty_handle.send_input(chars),
                timeout=self.state.timeouts.fast_op_s,
            )
            await asyncio.sleep(0.1)

        yield_time_ms = 250 if yield_time_s is None else int(yield_time_s * 1000)
        output, original_token_count = await self._collect_pty_output(
            entry=entry,
            yield_time_ms=resolve_pty_write_yield_time_ms(
                yield_time_ms=yield_time_ms, input_empty=chars == ""
            ),
            max_output_tokens=max_output_tokens,
        )
        entry.last_used = time.monotonic()
        return await self._finalize_pty_update(
            process_id=session_id,
            entry=entry,
            output=output,
            original_token_count=original_token_count,
        )

    async def _finalize_pty_update(
        self,
        *,
        process_id: int,
        entry: _DaytonaPtySessionEntry,
        output: bytes,
        original_token_count: int | None,
    ) -> PtyExecUpdate:
        exit_code = entry.exit_code if entry.done else None
        live_process_id: int | None = process_id

        if entry.done:
            async with self._pty_lock:
                removed = self._pty_sessions.pop(process_id, None)
                self._reserved_pty_process_ids.discard(process_id)
            if removed is not None:
                await self._terminate_pty_entry(removed)
            live_process_id = None

        return PtyExecUpdate(
            process_id=live_process_id,
            output=output,
            exit_code=exit_code,
            original_token_count=original_token_count,
        )

    async def pty_terminate_all(self) -> None:
        async with self._pty_lock:
            entries = list(self._pty_sessions.values())
            self._pty_sessions.clear()
            self._reserved_pty_process_ids.clear()
        for entry in entries:
            await self._terminate_pty_entry(entry)

    async def _collect_pty_output(
        self,
        *,
        entry: _DaytonaPtySessionEntry,
        yield_time_ms: int,
        max_output_tokens: int | None,
    ) -> tuple[bytes, int | None]:
        deadline = time.monotonic() + (yield_time_ms / 1000)
        output = bytearray()

        while True:
            async with entry.output_lock:
                while entry.output_chunks:
                    output.extend(entry.output_chunks.popleft())

            if time.monotonic() >= deadline:
                break

            if entry.done:
                async with entry.output_lock:
                    while entry.output_chunks:
                        output.extend(entry.output_chunks.popleft())
                break

            remaining_s = deadline - time.monotonic()
            if remaining_s <= 0:
                break

            try:
                await asyncio.wait_for(entry.output_notify.wait(), timeout=remaining_s)
            except asyncio.TimeoutError:
                break
            entry.output_notify.clear()

        text = output.decode("utf-8", errors="replace")
        truncated, original_token_count = truncate_text_by_tokens(text, max_output_tokens)
        return truncated.encode("utf-8", errors="replace"), original_token_count

    def _prune_pty_sessions_if_needed(self) -> _DaytonaPtySessionEntry | None:
        if len(self._pty_sessions) < PTY_PROCESSES_MAX:
            return None
        meta: list[tuple[int, float, bool]] = [
            (pid, entry.last_used, entry.done) for pid, entry in self._pty_sessions.items()
        ]
        pid = process_id_to_prune_from_meta(meta)
        if pid is None:
            return None
        self._reserved_pty_process_ids.discard(pid)
        return self._pty_sessions.pop(pid, None)

    async def _terminate_pty_entry(self, entry: _DaytonaPtySessionEntry) -> None:
        try:
            if entry.tty:
                await self._sandbox.process.kill_pty_session(entry.daytona_session_id)
            else:
                await self._sandbox.process.delete_session(entry.daytona_session_id)
        except Exception:
            pass

    async def read(self, path: Path | str, *, user: str | User | None = None) -> io.IOBase:
        error_path = posix_path_as_path(coerce_posix_path(path))
        if user is not None:
            workspace_path = await self._check_read_with_exec(path, user=user)
        else:
            workspace_path = await self._validate_path_access(path)

        daytona_exc = _import_daytona_exceptions()
        not_found_exc = daytona_exc.get("not_found")

        try:
            data: bytes = await self._sandbox.fs.download_file(
                sandbox_path_str(workspace_path),
                self.state.timeouts.file_download_s,
            )
            return io.BytesIO(data)
        except Exception as e:
            if not_found_exc is not None and isinstance(e, not_found_exc):
                raise WorkspaceReadNotFoundError(path=error_path, cause=e) from e
            raise WorkspaceArchiveReadError(path=error_path, cause=e) from e

    async def write(
        self,
        path: Path | str,
        data: io.IOBase,
        *,
        user: str | User | None = None,
    ) -> None:
        error_path = posix_path_as_path(coerce_posix_path(path))
        if user is not None:
            await self._check_write_with_exec(path, user=user)

        payload = data.read()
        if isinstance(payload, str):
            payload = payload.encode("utf-8")
        if not isinstance(payload, bytes | bytearray):
            raise WorkspaceWriteTypeError(path=error_path, actual_type=type(payload).__name__)

        workspace_path = await self._validate_path_access(path, for_write=True)
        try:
            await self._sandbox.fs.upload_file(
                bytes(payload),
                sandbox_path_str(workspace_path),
                timeout=self.state.timeouts.file_upload_s,
            )
        except Exception as e:
            raise WorkspaceArchiveWriteError(path=workspace_path, cause=e) from e

    async def running(self) -> bool:
        try:
            await asyncio.wait_for(
                self._sandbox.refresh_data(),
                timeout=self.state.timeouts.keepalive_s,
            )
            SandboxState = _import_sandbox_state()
            if SandboxState is None:
                return False
            return bool(getattr(self._sandbox, "state", None) == SandboxState.STARTED)
        except Exception:
            return False

    def _tar_exclude_args(self) -> list[str]:
        return shell_tar_exclude_args(self._persist_workspace_skip_relpaths())

    @retry_async(
        retry_if=lambda exc, self, tar_cmd, tar_path: (
            exception_chain_contains_type(exc, _retryable_persist_workspace_error_types())
            or exception_chain_has_status_code(exc, TRANSIENT_HTTP_STATUS_CODES)
        )
    )
    async def _run_persist_workspace_command(self, tar_cmd: str, tar_path: str) -> bytes:
        try:
            envs = await self._resolved_envs()
            result = await self._sandbox.process.exec(
                tar_cmd,
                env=envs or None,
                timeout=self.state.timeouts.workspace_tar_s,
            )
            if result.exit_code != 0:
                raise WorkspaceArchiveReadError(
                    path=self._workspace_root_path(),
                    context={"reason": "tar_failed", "output": result.result or ""},
                )
            return cast(
                bytes,
                await self._sandbox.fs.download_file(
                    tar_path,
                    self.state.timeouts.file_download_s,
                ),
            )
        except WorkspaceArchiveReadError:
            raise
        except Exception as e:
            raise WorkspaceArchiveReadError(path=self._workspace_root_path(), cause=e) from e

    async def persist_workspace(self) -> io.IOBase:
        def _error_context_summary(error: WorkspaceArchiveReadError) -> dict[str, str]:
            summary = {"message": error.message}
            if error.cause is not None:
                summary["cause_type"] = type(error.cause).__name__
                summary["cause"] = str(error.cause)
            return summary

        root = self._workspace_root_path()
        tar_path = f"/tmp/sandbox-persist-{self.state.session_id.hex}.tar"
        excludes = " ".join(self._tar_exclude_args())
        tar_cmd = (
            f"tar {excludes} -C {shlex.quote(root.as_posix())} -cf {shlex.quote(tar_path)} ."
        ).strip()

        unmounted_mounts: list[tuple[Mount, Path]] = []
        unmount_error: WorkspaceArchiveReadError | None = None
        for mount_entry, mount_path in self.state.manifest.ephemeral_mount_targets():
            try:
                await mount_entry.mount_strategy.teardown_for_snapshot(
                    mount_entry, self, mount_path
                )
            except Exception as e:
                unmount_error = WorkspaceArchiveReadError(path=root, cause=e)
                break
            unmounted_mounts.append((mount_entry, mount_path))

        snapshot_error: WorkspaceArchiveReadError | None = None
        raw: bytes | None = None
        if unmount_error is None:
            try:
                raw = await self._run_persist_workspace_command(tar_cmd, tar_path)
            except WorkspaceArchiveReadError as e:
                snapshot_error = e
            finally:
                try:
                    await self._sandbox.process.exec(
                        f"rm -f -- {shlex.quote(tar_path)}",
                        timeout=self.state.timeouts.cleanup_s,
                    )
                except Exception:
                    pass

        remount_error: WorkspaceArchiveReadError | None = None
        for mount_entry, mount_path in reversed(unmounted_mounts):
            try:
                await mount_entry.mount_strategy.restore_after_snapshot(
                    mount_entry, self, mount_path
                )
            except Exception as e:
                current_error = WorkspaceArchiveReadError(path=root, cause=e)
                if remount_error is None:
                    remount_error = current_error
                    if unmount_error is not None:
                        remount_error.context["earlier_unmount_error"] = _error_context_summary(
                            unmount_error
                        )
                else:
                    additional_remount_errors = remount_error.context.setdefault(
                        "additional_remount_errors",
                        [],
                    )
                    assert isinstance(additional_remount_errors, list)
                    additional_remount_errors.append(_error_context_summary(current_error))

        if remount_error is not None:
            if snapshot_error is not None:
                remount_error.context["snapshot_error_before_remount_corruption"] = (
                    _error_context_summary(snapshot_error)
                )
            raise remount_error
        if unmount_error is not None:
            raise unmount_error
        if snapshot_error is not None:
            raise snapshot_error

        assert raw is not None
        return io.BytesIO(raw)

    async def hydrate_workspace(self, data: io.IOBase) -> None:
        root = self._workspace_root_path()
        tar_path = f"/tmp/sandbox-hydrate-{self.state.session_id.hex}.tar"
        payload = data.read()
        if isinstance(payload, str):
            payload = payload.encode("utf-8")
        if not isinstance(payload, bytes | bytearray):
            raise WorkspaceWriteTypeError(path=Path(tar_path), actual_type=type(payload).__name__)

        try:
            validate_tar_bytes(
                bytes(payload),
                allow_external_symlink_targets=False,
            )
        except UnsafeTarMemberError as e:
            raise WorkspaceArchiveWriteError(
                path=root,
                context={
                    "reason": "unsafe_or_invalid_tar",
                    "member": e.member,
                    "detail": str(e),
                },
                cause=e,
            ) from e

        try:
            await self.mkdir(root, parents=True)
            envs = await self._resolved_envs()
            await self._sandbox.fs.upload_file(
                bytes(payload),
                tar_path,
                timeout=self.state.timeouts.file_upload_s,
            )
            result = await self._sandbox.process.exec(
                f"tar -C {shlex.quote(root.as_posix())} -xf {shlex.quote(tar_path)}",
                env=envs or None,
                timeout=self.state.timeouts.workspace_tar_s,
            )
            if result.exit_code != 0:
                raise WorkspaceArchiveWriteError(
                    path=root,
                    context={"reason": "tar_extract_failed", "output": result.result or ""},
                )
        except WorkspaceArchiveWriteError:
            raise
        except Exception as e:
            raise WorkspaceArchiveWriteError(path=root, cause=e) from e
        finally:
            try:
                envs = await self._resolved_envs()
                await self._sandbox.process.exec(
                    f"rm -f -- {shlex.quote(tar_path)}",
                    env=envs or None,
                    timeout=self.state.timeouts.cleanup_s,
                )
            except Exception:
                pass


class DaytonaSandboxClient(BaseSandboxClient[DaytonaSandboxClientOptions]):
    """Daytona sandbox client managing sandbox lifecycle via AsyncDaytona."""

    backend_id = "daytona"
    _instrumentation: Instrumentation

    def __init__(
        self,
        *,
        api_key: str | None = None,
        api_url: str | None = None,
        instrumentation: Instrumentation | None = None,
        dependencies: Dependencies | None = None,
    ) -> None:
        AsyncDaytona, DaytonaConfig, _, _ = _import_daytona_sdk()
        config = DaytonaConfig(api_key=api_key, api_url=api_url) if (api_key or api_url) else None
        self._daytona = AsyncDaytona(config)
        self._instrumentation = instrumentation or Instrumentation()
        self._dependencies = dependencies

    async def _build_create_params(
        self,
        *,
        sandbox_snapshot_name: str | None,
        image: str | None,
        env_vars: dict[str, str] | None,
        manifest: Manifest,
        name: str | None = None,
        resources: DaytonaSandboxResources | None = None,
        auto_stop_interval: int | None = None,
    ) -> Any:
        _, _, CreateSandboxFromSnapshotParams, CreateSandboxFromImageParams = _import_daytona_sdk()
        base_envs = dict(env_vars or {})
        creation_envs = base_envs or None

        if sandbox_snapshot_name:
            return CreateSandboxFromSnapshotParams(
                snapshot=sandbox_snapshot_name,
                env_vars=creation_envs,
                name=name,
                auto_stop_interval=auto_stop_interval,
            )

        if image:
            sandbox_resources = None
            if resources is not None and any(
                v is not None for v in (resources.cpu, resources.memory, resources.disk)
            ):
                Resources = _import_sdk_resources()
                sandbox_resources = Resources(
                    cpu=resources.cpu,
                    memory=resources.memory,
                    disk=resources.disk,
                )
            return CreateSandboxFromImageParams(
                image=image,
                env_vars=creation_envs,
                name=name,
                resources=sandbox_resources,
                auto_stop_interval=auto_stop_interval,
            )

        return CreateSandboxFromSnapshotParams(
            env_vars=creation_envs,
            name=name,
            auto_stop_interval=auto_stop_interval,
        )

    async def create(
        self,
        *,
        snapshot: SnapshotSpec | SnapshotBase | None = None,
        manifest: Manifest | None = None,
        options: DaytonaSandboxClientOptions,
    ) -> SandboxSession:
        if manifest is None:
            manifest = Manifest(root=DEFAULT_DAYTONA_WORKSPACE_ROOT)

        timeouts_in = options.timeouts
        if isinstance(timeouts_in, DaytonaSandboxTimeouts):
            timeouts = timeouts_in
        elif timeouts_in is None:
            timeouts = DaytonaSandboxTimeouts()
        else:
            timeouts = DaytonaSandboxTimeouts.model_validate(timeouts_in)

        session_id = uuid.uuid4()
        sandbox_name = options.name or str(session_id)

        params = await self._build_create_params(
            sandbox_snapshot_name=options.sandbox_snapshot_name,
            image=options.image,
            env_vars=options.env_vars,
            manifest=manifest,
            name=sandbox_name,
            resources=options.resources,
            auto_stop_interval=options.auto_stop_interval,
        )
        daytona_sandbox = await self._daytona.create(params, timeout=options.create_timeout)

        snapshot_instance = resolve_snapshot(snapshot, str(session_id))
        state = DaytonaSandboxSessionState(
            session_id=session_id,
            manifest=manifest,
            snapshot=snapshot_instance,
            sandbox_id=daytona_sandbox.id,
            sandbox_snapshot_name=options.sandbox_snapshot_name,
            image=options.image,
            base_env_vars=dict(options.env_vars or {}),
            pause_on_exit=options.pause_on_exit,
            create_timeout=options.create_timeout,
            start_timeout=options.start_timeout,
            name=sandbox_name,
            resources=options.resources,
            auto_stop_interval=options.auto_stop_interval,
            timeouts=timeouts,
            exposed_ports=options.exposed_ports,
            exposed_port_url_ttl_s=options.exposed_port_url_ttl_s,
        )
        inner = DaytonaSandboxSession.from_state(state, sandbox=daytona_sandbox)
        return self._wrap_session(inner, instrumentation=self._instrumentation)

    async def close(self) -> None:
        """Close the underlying AsyncDaytona HTTP client session."""
        await self._daytona.close()

    async def __aenter__(self) -> DaytonaSandboxClient:
        return self

    async def __aexit__(self, *_: object) -> None:
        await self.close()

    async def delete(self, session: SandboxSession) -> SandboxSession:
        inner = session._inner
        if not isinstance(inner, DaytonaSandboxSession):
            raise TypeError("DaytonaSandboxClient.delete expects a DaytonaSandboxSession")
        try:
            await inner.shutdown()
        except Exception:
            pass
        return session

    async def resume(
        self,
        state: SandboxSessionState,
    ) -> SandboxSession:
        if not isinstance(state, DaytonaSandboxSessionState):
            raise TypeError("DaytonaSandboxClient.resume expects a DaytonaSandboxSessionState")

        daytona_sandbox = None
        reconnected = False
        try:
            daytona_sandbox = await self._daytona.get(state.sandbox_id)
            SandboxState = _import_sandbox_state()
            if getattr(daytona_sandbox, "state", None) != SandboxState.STARTED:
                await daytona_sandbox.start(timeout=state.start_timeout)
            reconnected = True
        except Exception as e:
            logger.debug("daytona sandbox get() failed, will recreate: %s", e)

        if not reconnected or daytona_sandbox is None:
            params = await self._build_create_params(
                sandbox_snapshot_name=state.sandbox_snapshot_name,
                image=state.image,
                env_vars=state.base_env_vars,
                manifest=state.manifest,
                name=state.name,
                resources=state.resources,
                auto_stop_interval=state.auto_stop_interval,
            )
            daytona_sandbox = await self._daytona.create(params, timeout=state.create_timeout)
            state.sandbox_id = daytona_sandbox.id
            state.workspace_root_ready = False

        inner = DaytonaSandboxSession.from_state(state, sandbox=daytona_sandbox)
        inner._set_start_state_preserved(reconnected, system=reconnected)
        return self._wrap_session(inner, instrumentation=self._instrumentation)

    def deserialize_session_state(self, payload: dict[str, object]) -> SandboxSessionState:
        return DaytonaSandboxSessionState.model_validate(payload)


__all__ = [
    "DEFAULT_DAYTONA_WORKSPACE_ROOT",
    "DaytonaSandboxResources",
    "DaytonaSandboxClient",
    "DaytonaSandboxClientOptions",
    "DaytonaSandboxSession",
    "DaytonaSandboxSessionState",
    "DaytonaSandboxTimeouts",
]
