"""
CursorCLIAdapter — drives Cursor's `cursor-agent` CLI via per-turn subprocess.

Each turn spawns (headless, JSON event stream on stdout):
  cursor-agent -p --output-format stream-json --stream-partial-output \
    --force --model <model> --workspace <cwd> [--resume <chat_id>] "<text>"

`--model auto` is the default and is REQUIRED on Free plans (named models are
rejected); cursor-agent also otherwise falls back to a remembered named model,
so we always pass an explicit model. The chat/session id arrives in the
`system/init` event and is reused via `--resume <id>` to keep context.

stream-json events (one JSON object per line):
  {type:"system", subtype:"init", session_id, model, cwd, permissionMode}
  {type:"user", message}                                  ← our own echo (ignored)
  {type:"thinking", subtype:"delta", text}                ← streamed reasoning
  {type:"thinking", subtype:"completed"}
  {type:"assistant", message:{content:[{type:"text", text}]}}   ← a full text segment
  {type:"tool_call", subtype:"started"|"completed",
     call_id, tool_call:{<name>ToolCall:{args, result, description}}}
  {type:"interaction_query", ...}                         ← askQuestion/plan prompts (auto-handled headless)
  {type:"result", subtype:"success"|"error", is_error, result,
     usage:{inputTokens, outputTokens, cacheReadTokens, cacheWriteTokens}}

Note: `createPlanToolCall.args.plan` is a markdown string (no per-step status),
so — like gemini — there is no structured TodoBlock; it surfaces as a tool call.
"""
from __future__ import annotations
import asyncio
import json
import logging
import os
import shutil
import uuid
from pathlib import Path
from typing import AsyncIterator, Callable

from ..attachments import Attachment, build_text_with_inline
from ..blocks import (
    Block, AgentInfoBlock, AudioHintBlock, SessionEndBlock, SessionStatusHintBlock,
    TextBlock, ThinkingBlock, ToolCallBlock, tool_result_or_large,
)
from .base import (PerTurnCLIAdapter, ExternalSession, active_cwds_for_command, agent_env,
                   tool_label_subject)

log = logging.getLogger("beacon.cursor")

# Mirror the other per-turn adapters' active-turn window: transcript writes within this many
# seconds count as an in-progress turn; anything older is a quiet-but-open session.
_TURN_RECENT_SECS = 20.0

_CURSOR_PROJECTS = Path.home() / ".cursor" / "projects"
_CURSOR_MODELS_CACHE: tuple[float, list[dict]] | None = None
_CURSOR_MODELS_TTL_SECS = 900.0  # 15 min


def _decode_cursor_cwd(name: str) -> str | None:
    """Recover a real path from a cursor projects dir name (slashes→hyphens, no leading
    slash). Hyphens are ambiguous, so backtrack with filesystem checks (like claude's)."""
    tokens = name.split("-")
    if not tokens:
        return None

    def search(idx: int, parts: list[str]) -> str | None:
        # Ancestor prune (CRITICAL): all but the last of `parts` are finalized path
        # components; if their prefix isn't a real directory, no completion below can
        # exist — skip the subtree. Without this the hyphen-split backtracking is
        # O(2^hyphens) and a hyphen-heavy / deleted dir name (e.g. a stale pytest temp
        # dir) explodes, blocking the event loop on filesystem stats. Mirrors the fix in
        # claude_code._decode_project_name — see memory project_adapter_pathdecode_cpu_spin.
        if len(parts) > 1:
            parent = "/" + "/".join(parts[:-1])
            if not os.path.isdir(parent):
                return None
        if idx == len(tokens):
            p = "/" + "/".join(parts)
            return p if os.path.isdir(p) else None
        tok = tokens[idx]
        r = search(idx + 1, parts + [tok])
        if r:
            return r
        if parts:
            r = search(idx + 1, parts[:-1] + [parts[-1] + "-" + tok])
            if r:
                return r
        return None

    return search(0, [])


def _cursor_first_user(path) -> str:
    try:
        with path.open("r", errors="replace") as f:
            for _ in range(40):
                line = f.readline()
                if not line:
                    break
                try:
                    o = json.loads(line)
                except Exception:
                    continue
                if o.get("role") == "user":
                    for c in (o.get("message", {}).get("content") or []):
                        if isinstance(c, dict) and c.get("type") == "text" and c.get("text"):
                            t = c["text"].strip()
                            if t.startswith("<user_query>"):
                                t = t[len("<user_query>"):].rsplit("</user_query>", 1)[0].strip()
                            return t[:200]
    except Exception:
        pass
    return ""


def _scan_cursor_sessions(limit: int = 20) -> list[ExternalSession]:
    if not _CURSOR_PROJECTS.exists():
        return []
    import time as _t
    owned = {os.path.realpath(c) for c in active_cwds_for_command("cursor-agent")}
    cands: list[tuple[float, "Path", str | None]] = []
    for proj in _CURSOR_PROJECTS.iterdir():
        at = proj / "agent-transcripts"
        if not at.is_dir():
            continue
        cwd = _decode_cursor_cwd(proj.name)
        for chatdir in at.iterdir():
            f = chatdir / f"{chatdir.name}.jsonl"
            if not f.exists():
                continue
            try:
                cands.append((f.stat().st_mtime, f, cwd))
            except OSError:
                continue
    cands.sort(key=lambda x: x[0], reverse=True)
    now = _t.time()
    out: list[ExternalSession] = []
    for mt, f, cwd in cands[:limit]:
        is_active = bool(cwd) and os.path.realpath(cwd) in owned
        activity = ("turn" if (now - mt) <= _TURN_RECENT_SECS else "open") if is_active else "idle"
        out.append(ExternalSession(session_id=f.stem, agent_id="cursor", cwd=cwd or "",
                                   last_prompt=_cursor_first_user(f), started_at="",
                                   is_active=is_active, activity=activity))
    return out


class CursorCLIAdapter(PerTurnCLIAdapter):
    cli_name = "cursor-agent"
    cap_images = False           # cursor-agent image input unconfirmed → off until verified
    cap_documents = False
    # P4: cursor loads the beacon MCP, but headless `-p` needs --approve-mcps to execute the
    # call, and cursor sanitizes the MCP child env (per-session routing comes from ${VAR}
    # expansion in ~/.cursor/mcp.json). With both in place, delivery is verified end-to-end.
    cap_sentinel = True
    cap_thinking = True          # emits thinking/delta + thinking/completed
    cap_interrupt = True         # interrupt = kill the subprocess
    cap_input_during_turn = False
    provides_models = True
    provides_usage = True
    agent_id = "cursor"
    display_name = "Cursor"
    beta = False              # live-verified via the adapter matrix
    cli_binary = "cursor-agent"

    def __init__(
        self,
        binary: str | None = None,
        cwd: str | None = None,
        model: str | None = None,
        resume_session_id: str | None = None,
        on_session_id: "Callable[[str], None] | None" = None,
    ) -> None:
        self._binary = binary or shutil.which("cursor-agent") or "cursor-agent"
        self._cwd = cwd
        self._model = model or "auto"           # Free plans / remembered-model gotcha → always explicit
        self._chat_id: str | None = resume_session_id   # reused via --resume to keep context
        self._on_session_id = on_session_id     # persists the cursor chat id for handoff dedup
        self._send_queue: asyncio.Queue[tuple[str, list[Attachment]]] = None  # type: ignore
        self._proc: asyncio.subprocess.Process | None = None
        self._fetch_store: dict[str, str] = {}
        self._thinking_buf = ""
        self._text_buf = ""
        self._text_id: str | None = None
        self._usage: dict = {}

    # ── Session handoff (docs/session-handoff.md) ──────────────────────────────

    @classmethod
    def scan_external_sessions(cls) -> list[ExternalSession]:
        return _scan_cursor_sessions()

    @classmethod
    def is_cwd_externally_owned(cls, cwd: str, sid: str = "") -> bool:
        if not cwd:
            return False
        target = os.path.realpath(os.path.expanduser(cwd))
        return any(os.path.realpath(c) == target for c in active_cwds_for_command("cursor-agent"))

    @classmethod
    def latest_sid_for_cwd(cls, cwd: str) -> str | None:
        if not cwd:
            return None
        enc = os.path.expanduser(cwd).lstrip("/").replace("/", "-")
        at = _CURSOR_PROJECTS / enc / "agent-transcripts"
        if not at.is_dir():
            return None
        newest: tuple[float, str] | None = None
        for chatdir in at.iterdir():
            f = chatdir / f"{chatdir.name}.jsonl"
            if not f.exists():
                continue
            try:
                mt = f.stat().st_mtime
            except OSError:
                continue
            if newest is None or mt > newest[0]:
                newest = (mt, chatdir.name)
        return newest[1] if newest else None

    @classmethod
    def make_transcript_mirror(cls, cwd: str, sid: str, *, from_end: bool = False):
        from ..transcript_mirror import make_cursor_mirror
        return make_cursor_mirror(sid, from_end=from_end)

    @classmethod
    def terminal_resume_command(cls, sid: str) -> str | None:
        import shlex
        return f"cursor-agent --resume {shlex.quote(sid)}" if sid else "cursor-agent"

    @classmethod
    def readiness(cls) -> tuple[bool, str]:
        if (Path.home() / ".cursor" / "cli-config.json").exists():
            return True, ""
        return False, "Run `cursor-agent login`."

    # ── Lifecycle ─────────────────────────────────────────────────────────────

    async def start(self, session_id: str) -> AgentInfoBlock:
        self._beacon_session_id = session_id   # stamped into per-turn spawn env (beacon MCP routing)
        self._send_queue = asyncio.Queue()
        return self.make_agent_info(
            name="Cursor",
            working_dirs=[self._cwd] if self._cwd else [],
        )

    def clear_history(self) -> None:
        # Leaseless: drop the cursor chat id so the next per-turn spawn omits --resume
        # (cursor mints a fresh chat id). Wipe the persisted ext_sid too so a reconnect
        # doesn't resume the old chat.
        self._chat_id = None
        if self._on_session_id:
            self._on_session_id("")

    async def get_usage(self) -> dict:
        return dict(self._usage)

    @classmethod
    async def list_models(cls, extra: dict | None = None) -> list[dict]:
        """`cursor-agent models` → [{id, label, detail}]. Lines are 'id - Label'."""
        import time as _time
        global _CURSOR_MODELS_CACHE
        now = _time.monotonic()
        if _CURSOR_MODELS_CACHE is not None:
            cached_ts, cached = _CURSOR_MODELS_CACHE
            if cached and (now - cached_ts) < _CURSOR_MODELS_TTL_SECS:
                return cached

        binary = shutil.which("cursor-agent") or "cursor-agent"
        try:
            proc = await asyncio.create_subprocess_exec(
                binary, "models",
                stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL,
                env=agent_env(),
            )
            out, _ = await asyncio.wait_for(proc.communicate(), timeout=10)
        except Exception:
            return []
        models: list[dict] = []
        for line in out.decode().splitlines():
            line = line.strip()
            if not line or " - " not in line:
                continue
            mid, _, label = line.partition(" - ")
            mid, label = mid.strip(), label.strip()
            if mid and " " not in mid:   # skip prose lines ("Available models")
                models.append({"id": mid, "label": label or mid, "detail": ""})
        _CURSOR_MODELS_CACHE = (now, models)
        return models

    # ── Turn loop (spawn/stream/teardown handled by PerTurnCLIAdapter) ─────────

    async def _build_argv(self, text: str, attachments: list[Attachment]) -> list[str]:
        cmd = [
            self._binary, "-p",
            "--output-format", "stream-json",
            "--stream-partial-output",
            "--force",
            "--model", self._model,
        ]
        # Beacon sentinel tools: cursor loads the globally-registered beacon MCP, but in
        # headless `-p` it requires --approve-mcps or the MCP tool call is never executed.
        # Routing (port + session) reaches the sanitized MCP child via ${VAR} env expansion
        # in ~/.cursor/mcp.json (written by setup_integrations._setup_cursor).
        if type(self).cap_sentinel:
            cmd.append("--approve-mcps")
        if self._cwd:
            cmd += ["--workspace", self._cwd]
        if self._chat_id:
            cmd += ["--resume", self._chat_id]
        # cursor-agent has no native image input → inline attachments as text (Mac-OCR for
        # images + path hints), so an attached photo still reaches the model. See attachments.py.
        preamble = self.sentinel_client_preamble(is_first_turn=not self._chat_id)
        cmd.append(preamble + await asyncio.to_thread(build_text_with_inline, text, attachments))
        return cmd

    def _pre_spawn(self) -> None:
        # Keep ~/.terminull/sentinel_mcp.py current before each spawn (TOOL_DEFS may change).
        if type(self).cap_sentinel:
            try:
                from ..sentinel_tools import write_global_script
                write_global_script()
            except Exception:
                log.debug("cursor: sentinel write_global_script failed", exc_info=True)

    def _reset_turn_state(self) -> None:
        self._thinking_buf = ""
        self._text_buf = ""
        self._text_id = None

    # ── Event parser ──────────────────────────────────────────────────────────

    async def _parse_event(self, event: dict) -> AsyncIterator[Block]:
        etype = event.get("type", "")
        subtype = event.get("subtype", "")

        if etype == "system" and subtype == "init":
            # Capture the chat id for --resume on later turns; refresh the header.
            if sid := event.get("session_id"):
                if sid != self._chat_id and self._on_session_id:
                    self._on_session_id(sid)
                self._chat_id = sid
            yield self.make_agent_info(
                name="Cursor",
                version=str(event.get("model", "")),
                working_dirs=[self._cwd] if self._cwd else [],
            )

        elif etype == "thinking":
            if subtype == "delta":
                self._thinking_buf += event.get("text", "")
            elif subtype == "completed":
                if self._thinking_buf.strip():
                    yield ThinkingBlock(thinking=self._thinking_buf)
                self._thinking_buf = ""

        elif etype == "assistant":
            # Cursor usually emits full assistant segments, but with
            # `--stream-partial-output` it may also emit delta/completed events.
            if subtype == "delta":
                chunk = event.get("text", "") or ""
                if chunk:
                    if self._text_id is None:
                        self._text_id = str(uuid.uuid4())
                    self._text_buf += chunk
                    yield TextBlock(text=chunk, done=False, id=self._text_id)
            elif subtype == "completed":
                if self._text_id is not None and self._text_buf:
                    yield TextBlock(text=self._text_buf, done=True, id=self._text_id)
                self._text_buf = ""
                self._text_id = None
            else:
                text = "".join(
                    c.get("text", "")
                    for c in event.get("message", {}).get("content", [])
                    if c.get("type") == "text"
                )
                if text:
                    # Snapshot-only path: emit one completed block with a fresh id.
                    yield TextBlock(text=text, done=True, id=str(uuid.uuid4()))

        elif etype == "tool_call":
            name, args, inner = _tool_fields(event)
            call_id = event.get("call_id") or str(uuid.uuid4())
            if subtype == "started":
                lbl, subj = tool_label_subject(name, args)
                yield AudioHintBlock(state="agent_silent")
                yield ToolCallBlock(id=call_id, name=name, args=args,
                                    status="running", label=lbl, subject=subj)
            elif subtype == "completed":
                lbl, subj = tool_label_subject(name, args)
                content, exit_code, is_error = _tool_result(inner)
                yield ToolCallBlock(id=call_id, name=name, args=args,
                                    status="error" if is_error else "done",
                                    label=lbl, subject=subj)
                if content:
                    yield tool_result_or_large(
                        tool_call_id=call_id, content=content,
                        fetch_store=self._fetch_store,
                        exit_code=exit_code, is_error=is_error,
                    )
                yield AudioHintBlock(state="agent_speaking")

        elif etype == "result":
            usage = event.get("usage") or {}
            self._usage = {
                "input_tokens": usage.get("inputTokens", 0),
                "output_tokens": usage.get("outputTokens", 0),
                "cache_read_tokens": usage.get("cacheReadTokens", 0),
                "cache_write_tokens": usage.get("cacheWriteTokens", 0),
            }
            yield SessionStatusHintBlock(
                model=self._model,
                output_tokens=self._usage["output_tokens"],
                last_input_tokens=self._usage["input_tokens"],
                cache_creation_tokens=self._usage["cache_write_tokens"],
            )
            yield AudioHintBlock(state="agent_silent")
            yield SessionEndBlock(reason="error" if event.get("is_error") else "completed")

        # `interaction_query` (askQuestion / createPlan prompts) auto-resolve in
        # headless --force mode → nothing to surface. `user` is our own echo.


# ── Helpers ───────────────────────────────────────────────────────────────────

def _tool_fields(event: dict) -> tuple[str, dict, dict]:
    """Return (tool_name, args, inner) from a tool_call event.

    The tool is nested under a single `<name>ToolCall` key, e.g. shellToolCall.
    """
    tc = event.get("tool_call") or {}
    key = next(iter(tc.keys()), "")
    inner = tc.get(key) or {}
    name = key[:-len("ToolCall")] if key.endswith("ToolCall") else (key or "tool")
    return name, (inner.get("args") or {}), inner


def _tool_result(inner: dict) -> tuple[str, int | None, bool]:
    """Extract (content, exit_code, is_error) from a completed tool_call inner."""
    result = inner.get("result")
    if isinstance(result, dict):
        success = result.get("success")
        if isinstance(success, dict) and "stdout" in success:
            out = success.get("stdout", "")
            if success.get("stderr"):
                out = (out + "\n" + success["stderr"]).strip()
            return out, success.get("exitCode"), False
        if "error" in result or "failure" in result:
            return json.dumps(result.get("error") or result.get("failure")), None, True
        return json.dumps(result), None, False
    if result is None:
        return "", None, False
    return str(result), None, False
