"""HarnessBench adapter for OpenDesktop (apps/server driven headlessly over HTTP).

Canonical copy lives in the OpenDesktop repo at evals/harnessbench/; install.sh
copies it into a harness-bench clone as src/harnessbench/adapters/opendesktop.py
and registers it. Audited harness-bench commit: 1025086a446653702b80cfb48babbeec35db6b2c.

Flow per run() call (one prompt = one turn):
  1. Register a proxy route so OPENDESKTOP_GATEWAY_URL points at the bench
     usage-proxy, which forwards to the real upstream and records tokens.
  2. Boot opendesktop-server with an isolated data dir inside the sandbox and
     cwd = the task workspace. Wait for /health.
  3. Create (or resume, for multi-prompt tasks) a session anchored to the
     workspace, set permission-mode=auto, subscribe to SSE, send the prompt.
  4. Wait for session_idle, collect assistant text + usage, shut the server down.
"""

from __future__ import annotations

import json
import os
import queue
import signal
import socket
import subprocess
import threading
import time
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any

from harnessbench.adapters.base import BaseAdapter
from harnessbench.models import AdapterRunContext, AdapterRunResult
from harnessbench.usage_proxy import register_routes

PROXY_PREFIX = "/opendesktop/gateway"
DEFAULT_UPSTREAM = "https://openrouter.ai/api/v1"
HEALTH_TIMEOUT_SEC = 60
BUSY_TIMEOUT_SEC = 30

# The system prompt tells the agent to call ask_user when a deliverable's
# format is unnamed. Headless, nobody answers — the turn would hang until the
# task timeout and score ~0. Auto-answer every question with this instead.
QUESTION_AUTO_ANSWER = (
    "Use your best judgment; produce the deliverable exactly as the task specifies."
)


def _free_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


def _http(method: str, url: str, body: dict[str, Any] | None = None, timeout: int = 30) -> tuple[int, dict[str, Any] | None]:
    data = json.dumps(body).encode("utf-8") if body is not None else None
    req = urllib.request.Request(url, data=data, method=method)
    if data is not None:
        req.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read()
            try:
                return resp.status, json.loads(raw) if raw else None
            except json.JSONDecodeError:
                return resp.status, None
    except urllib.error.HTTPError as exc:
        return exc.code, None
    except urllib.error.URLError:
        return 0, None


class _SseReader(threading.Thread):
    """Reads SSE events from /sessions/{id}/events into a queue."""

    def __init__(self, url: str, timeout_sec: int) -> None:
        super().__init__(daemon=True)
        self.url = url
        self.timeout_sec = timeout_sec
        self.events: queue.Queue[dict[str, Any]] = queue.Queue()
        self.error: str | None = None
        self._stop = threading.Event()

    def stop(self) -> None:
        self._stop.set()

    def run(self) -> None:
        try:
            req = urllib.request.Request(self.url, headers={"Accept": "text/event-stream"})
            with urllib.request.urlopen(req, timeout=self.timeout_sec) as resp:
                for raw_line in resp:
                    if self._stop.is_set():
                        return
                    line = raw_line.decode("utf-8", errors="replace").strip()
                    if not line.startswith("data:"):
                        continue
                    payload = line[5:].strip()
                    if not payload:
                        continue
                    try:
                        event = json.loads(payload)
                    except json.JSONDecodeError:
                        continue
                    if isinstance(event, dict):
                        self.events.put(event)
        except Exception as exc:  # connection drop, timeout, server exit
            self.error = str(exc)


class OpenDesktopAdapter(BaseAdapter):
    name = "opendesktop"

    def run(self, ctx: AdapterRunContext) -> AdapterRunResult:
        command = str(ctx.model_config.get("command") or "opendesktop-server")
        model = str(ctx.model_config.get("model") or "")
        api_key = str(ctx.model_config.get("api_key") or os.environ.get("OPENROUTER_API_KEY") or "")
        upstream = str(ctx.model_config.get("upstream_base_url") or DEFAULT_UPSTREAM)

        od_dir = ctx.sandbox / ".opendesktop"
        data_dir = od_dir / "data"
        data_dir.mkdir(parents=True, exist_ok=True)
        server_log = od_dir / "server.log"
        session_marker = od_dir / "session_id"

        # Route LLM traffic through the bench usage-proxy when available.
        proxy_base = str(ctx.env.get("HARNESSBENCH_LLM_PROXY_URL") or "").rstrip("/")
        routes_file = ctx.env.get("HARNESSBENCH_LLM_PROXY_ROUTES")
        if proxy_base and routes_file:
            register_routes(
                Path(routes_file),
                {PROXY_PREFIX: {"framework": "opendesktop", "provider": "gateway", "upstream": upstream}},
            )
            gateway_url = f"{proxy_base}{PROXY_PREFIX}"
        else:
            gateway_url = upstream

        port = _free_port()
        base = f"http://127.0.0.1:{port}"

        env = os.environ.copy()
        env.update(ctx.env)
        env["HOME"] = str(ctx.sandbox)
        env["OPENDESKTOP_DATA_DIR"] = str(data_dir)
        env["OPENDESKTOP_CWD"] = str(ctx.workspace.resolve())
        env["OPENDESKTOP_PORT"] = str(port)
        env["OPENDESKTOP_HOST"] = "127.0.0.1"
        env["OPENDESKTOP_GATEWAY_URL"] = gateway_url
        env["OPENROUTER_API_KEY"] = api_key
        if model:
            env["OPENDESKTOP_MODEL"] = model
        # Local/Enthusiast mode, open localhost API, no workspace config auto-trust.
        for key in ("OPENDESKTOP_CONTROL_PLANE_URL", "OPENDESKTOP_API_TOKEN", "OPENDESKTOP_TRUST_WORKSPACE"):
            env.pop(key, None)

        deadline = time.monotonic() + ctx.timeout_sec
        log_handle = server_log.open("a", encoding="utf-8")
        proc = subprocess.Popen(
            [command],
            cwd=str(ctx.sandbox),
            env=env,
            stdout=log_handle,
            stderr=subprocess.STDOUT,
            text=True,
        )
        try:
            result = self._drive(ctx, base, session_marker, proc, deadline)
        finally:
            self._shutdown(proc)
            log_handle.close()
        result.metadata.update(
            {
                "server_log": str(server_log),
                "data_dir": str(data_dir),
                "state_dir": str(data_dir),
                "port": port,
                "gateway_url": gateway_url,
                "workspace": str(ctx.workspace),
            }
        )
        return result

    def _drive(
        self,
        ctx: AdapterRunContext,
        base: str,
        session_marker: Path,
        proc: subprocess.Popen,
        deadline: float,
    ) -> AdapterRunResult:
        # 1. Health.
        health_deadline = min(deadline, time.monotonic() + HEALTH_TIMEOUT_SEC)
        while True:
            if proc.poll() is not None:
                return AdapterRunResult(ok=False, stderr=f"opendesktop-server exited early (code {proc.returncode}); see server.log")
            status, _ = _http("GET", f"{base}/health", timeout=2)
            if status == 200:
                break
            if time.monotonic() > health_deadline:
                return AdapterRunResult(ok=False, stderr="opendesktop-server /health never became ready")
            time.sleep(0.25)

        # 2. Create or resume the session (multi-prompt tasks reuse the marker).
        session_id = session_marker.read_text(encoding="utf-8").strip() if session_marker.is_file() else ""
        if not session_id:
            status, body = _http("POST", f"{base}/sessions", {"cwd": str(ctx.workspace.resolve())})
            if status not in (200, 201) or not body or not body.get("id"):
                return AdapterRunResult(ok=False, stderr=f"create session failed (HTTP {status})")
            session_id = str(body["id"])
            session_marker.write_text(session_id, encoding="utf-8")
        _http("POST", f"{base}/sessions/{session_id}/permission-mode", {"mode": "auto"})

        # 3. Subscribe, then send.
        remaining = max(5, int(deadline - time.monotonic()))
        reader = _SseReader(f"{base}/sessions/{session_id}/events", timeout_sec=remaining)
        reader.start()
        time.sleep(0.3)  # let the SSE subscription land before the turn starts
        status, _ = _http("POST", f"{base}/sessions/{session_id}/messages", {"text": ctx.prompt})
        if status != 202:
            reader.stop()
            return AdapterRunResult(ok=False, stderr=f"send message failed (HTTP {status})")

        # 4. Consume events until session_idle after the turn started.
        saw_busy = False
        assistant_text = ""
        usage_totals: dict[str, int] = {}
        errors: list[str] = []
        notes: list[str] = []
        busy_deadline = time.monotonic() + BUSY_TIMEOUT_SEC
        while True:
            now = time.monotonic()
            if now > deadline:
                _http("POST", f"{base}/sessions/{session_id}/abort", {})
                reader.stop()
                return AdapterRunResult(
                    ok=False,
                    stdout=assistant_text,
                    stderr=f"timeout after {ctx.timeout_sec}s waiting for session_idle",
                    metadata={"session_id": session_id, "usage": usage_totals, "turn_errors": errors},
                )
            if reader.error and proc.poll() is not None:
                return AdapterRunResult(ok=False, stdout=assistant_text, stderr=f"server died mid-turn: {reader.error}")
            try:
                event = reader.events.get(timeout=1.0)
            except queue.Empty:
                if not saw_busy and now > busy_deadline:
                    reader.stop()
                    return AdapterRunResult(ok=False, stderr="turn never started (no session_busy within 30s)")
                continue

            etype = event.get("type")
            if etype == "session_busy":
                saw_busy = True
            elif etype == "question":
                request_id = str(event.get("request_id") or "")
                if request_id:
                    _http(
                        "POST",
                        f"{base}/sessions/{session_id}/questions/{request_id}",
                        {"answer": QUESTION_AUTO_ANSWER},
                    )
                    notes.append(f"auto-answered ask_user: {event.get('question')!r}")
            elif etype == "message_complete" and event.get("role") == "assistant":
                assistant_text = str(event.get("text") or assistant_text)
            elif etype == "usage":
                for k in ("input_tokens", "output_tokens", "total_tokens"):
                    if isinstance(event.get(k), int):
                        usage_totals[k] = usage_totals.get(k, 0) + event[k]
            elif etype == "error":
                errors.append(str(event.get("message") or ""))
            elif etype == "session_idle" and saw_busy:
                reader.stop()
                # A turn that ended with error events is an infrastructure /
                # provider failure, not a graded outcome: the oracle still
                # scores whatever partial work exists, but the result must be
                # visibly marked so cross-model boards can exclude or flag it
                # instead of reading it as harness quality.
                return AdapterRunResult(
                    ok=True,
                    stdout=assistant_text,
                    stderr="\n".join(errors),
                    metadata={
                        "session_id": session_id,
                        "usage": usage_totals,
                        "turn_errors": errors,
                        "adapter_notes": notes,
                        "terminal_error": bool(errors),
                    },
                )

    @staticmethod
    def _shutdown(proc: subprocess.Popen) -> None:
        if proc.poll() is not None:
            return
        proc.send_signal(signal.SIGTERM)
        try:
            proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            proc.kill()
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                pass