"""Daytona-backed code sandbox — runs model-generated tests in a Daytona cloud sandbox.

WHY THIS EXISTS (read with orchestrator/sandbox.py security model and
docs/integration/security-boundary.md §8.1):
  The default backend executes generated code in a child process **inside this pod**. Daytona
  runs it in a **dedicated external sandbox** (its own isolated VM/container) — exactly the
  "专用 sandbox worker/job" evolution path §8.1 names. Code never touches this pod's process
  space, and we send ONLY the generated files + the test runner (never our env / secrets).

GATING IS UNCHANGED. This module is reached only from sandbox.run_tests(), which already called
  assert_isolated() (HEICODE_SANDBOX_ISOLATED) under the ENABLE_QUALITY_EVAL feature gate. Picking
  the Daytona backend does NOT weaken the fail-closed dual gate — it only changes where code runs.

CONTRACT: run_tests_daytona() returns the SAME SandboxResult shape as the local backend, by
  reusing sandbox_runner.py verbatim inside the Daytona sandbox and reading its _result.json.
  Raises on any Daytona/transport error so the caller can fail-soft to the local backend.
"""
from __future__ import annotations

import json
import logging
import os
from pathlib import Path
from typing import List

from .sandbox import SandboxFile, SandboxResult, DEFAULT_TIMEOUT_SECONDS

logger = logging.getLogger(__name__)

_RUNNER = Path(__file__).resolve().parent / "sandbox_runner.py"
_SBX_SUBDIR = "swarm_sbx"
_RESULT_SENTINEL = "___SBX_RESULT_JSON___"
_OUTPUT_CAP = 16 * 1024


def daytona_backend_selected() -> bool:
    return os.getenv("SANDBOX_BACKEND", "local").strip().lower() == "daytona"


def _safe_rel(path: str) -> str:
    """Reject absolute paths / `..` escape — same guard as the local backend's _safe_join."""
    if not path or os.path.isabs(path):
        raise ValueError(f"unsafe file path: {path!r}")
    norm = os.path.normpath(path)
    if norm.startswith("..") or norm.startswith("/"):
        raise ValueError(f"path escapes sandbox: {path!r}")
    return norm


def _client():
    """Build a Daytona client from env. Raises if the SDK is absent or no api_key is set."""
    from daytona import Daytona, DaytonaConfig  # lazy: only when daytona backend is on

    api_key = os.getenv("DAYTONA_API_KEY")
    if not api_key:
        raise RuntimeError("DAYTONA_API_KEY is not set")
    cfg_kwargs = {"api_key": api_key}
    api_url = os.getenv("DAYTONA_API_URL")
    if api_url:
        cfg_kwargs["api_url"] = api_url
    target = os.getenv("DAYTONA_TARGET")
    if target:
        cfg_kwargs["target"] = target
    return Daytona(DaytonaConfig(**cfg_kwargs))


def _parse_result(stdout: str) -> SandboxResult:
    """Extract the runner's JSON (after the sentinel) and map it to a SandboxResult."""
    result = SandboxResult()
    result.stdout = (stdout or "")[:_OUTPUT_CAP]
    if not stdout or _RESULT_SENTINEL not in stdout:
        result.error = "no result produced by daytona sandbox runner"
        return result
    raw = stdout.rsplit(_RESULT_SENTINEL, 1)[1].strip()
    try:
        data = json.loads(raw)
    except Exception as exc:
        result.error = f"unparseable daytona result: {exc!r}"
        return result
    result.total = int(data.get("total", 0))
    result.passed = int(data.get("passed", 0))
    result.failed = int(data.get("failed", 0))
    result.errored = int(data.get("errored", 0))
    result.details = data.get("details", []) or []
    if data.get("fatal"):
        result.error = str(data["fatal"])
    return result


def run_tests_daytona(
    source_files: List[SandboxFile],
    test_files: List[SandboxFile],
    *,
    timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
) -> SandboxResult:
    """Run source + test files in a Daytona cloud sandbox; return a SandboxResult.

    Uploads the generated files + sandbox_runner.py into an ephemeral dir in a freshly-created
    Daytona sandbox, runs the runner (pytest, stdlib fallback), and reads its _result.json. The
    sandbox is always deleted. Raises on Daytona/transport failure (caller fails soft to local).
    """
    from daytona import FileUpload  # lazy

    runner_src = _RUNNER.read_text(encoding="utf-8")
    exec_timeout = int(timeout_seconds) + 120  # SDK exec timeout: test wall-clock + setup overhead

    daytona = _client()
    sandbox = daytona.create()
    try:
        root = (sandbox.get_user_root_dir() or "/home/daytona").rstrip("/")
        workdir = f"{root}/{_SBX_SUBDIR}"

        uploads = [FileUpload(source=runner_src.encode("utf-8"), destination=f"{workdir}/_runner.py")]
        for f in list(source_files) + list(test_files):
            rel = _safe_rel(f.path)
            uploads.append(FileUpload(source=f.content.encode("utf-8"),
                                      destination=f"{workdir}/{rel}"))
        sandbox.fs.upload_files(uploads)

        # Best-effort pytest install; runner falls back to stdlib if it's unavailable. Quiet, and
        # never fatal — a non-zero pip exit still lets the runner produce a result.
        try:
            sandbox.process.exec("python -m pip install -q pytest 2>/dev/null || true",
                                 timeout=exec_timeout)
        except Exception as exc:  # pragma: no cover - network dependent
            logger.info("daytona: pytest install best-effort failed (runner will fall back): %s", exc)

        # Run the runner, then print _result.json after a sentinel so we can parse it out of the
        # combined stdout/stderr stream (we never trust loose stdout for counts).
        cmd = (f"cd {workdir} && python -I _runner.py >/dev/null 2>&1; "
               f"echo {_RESULT_SENTINEL}; cat _result.json")
        resp = sandbox.process.exec(cmd, timeout=exec_timeout)
        result = _parse_result(getattr(resp, "result", "") or "")
        if result.exit_code is None:
            result.exit_code = getattr(resp, "exit_code", None)
        return result
    finally:
        try:
            sandbox.delete()
        except Exception as exc:  # pragma: no cover - cleanup best-effort
            logger.warning("daytona: sandbox delete failed (may leak a sandbox): %s", exc)