"""
DroidAdapter — drives Factory.ai's `droid` CLI via per-turn subprocess.

Each turn spawns (headless, JSON event stream on stdout):
  droid exec --output-format stream-json --auto <level> \
    [--session-id <id>] [--model <model>] "<text>"

`droid exec` is the non-interactive mode (the analog of `claude -p` /
`cursor-agent -p`). `session_id` (present on every event) is reused via
`--session-id <id>` to keep context across turns (`--fork <id>` branches — unused).

stream-json events (one JSON object per line) — confirmed from a live capture:
  {type:"system", subtype:"init", session_id, cwd, model, reasoning_effort, tools:[…]}
  {type:"message", role:"user", text, id, session_id}        ← our own echo (ignored)
  {type:"message", role:"assistant", text, id, session_id}   ← a full assistant text segment
  {type:"tool_call", id, toolName, toolId, parameters, messageId, session_id}
  {type:"tool_result", id, toolId, isError, value, messageId, session_id}  ← id matches tool_call
  {type:"completion", finalText, numTurns, durationMs, usage:{input_tokens,
     output_tokens, cache_read_input_tokens, cache_creation_input_tokens}, session_id}
  {type:"error", source, message, session_id}                ← e.g. 402 no-subscription

Auth: FACTORY_API_KEY / Factory subscription, or BYOK custom models
(`--model custom:<id>`, configured in ~/.factory/settings.json) — inherited via
agent_env(). Autonomy: --auto high acts without per-tool prompts, matching the
other adapters' "user already trusted this agent" posture (claude
skip_permissions / cursor --force). Tunable via the `autonomy` ctor arg.

Note: no separate thinking/reasoning event was observed (cap_thinking stays
False until one is confirmed for a reasoning model). `value` on a tool_result
can be a string or structured — coerced to text for the result block.
"""
from __future__ import annotations
import asyncio
import json
import logging
import os
import re
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, ErrorBlock, SessionEndBlock,
    SessionStatusHintBlock, TextBlock, ToolCallBlock, tool_result_or_large,
)
from .base import (PerTurnCLIAdapter, ExternalSession, ORPolicy, active_cwds_for_command,
                   agent_env, tool_label_subject)

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

# 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

_FACTORY_SESSIONS = Path.home() / ".factory" / "sessions"


def _droid_session_head(path: Path) -> tuple[str, str]:
    """(cwd, title) from a droid session's line-1 `session_start` record."""
    try:
        with path.open("r", errors="replace") as f:
            o = json.loads(f.readline())
        if o.get("type") == "session_start":
            return o.get("cwd", "") or "", (o.get("title") or o.get("sessionTitle") or "")
    except Exception:
        pass
    return "", ""


def _scan_droid_sessions(limit: int = 20) -> list[ExternalSession]:
    if not _FACTORY_SESSIONS.exists():
        return []
    import time as _t
    owned = {os.path.realpath(c) for c in active_cwds_for_command("/droid")}
    cands: list[tuple[float, Path]] = []
    for f in _FACTORY_SESSIONS.glob("*/*.jsonl"):
        try:
            cands.append((f.stat().st_mtime, f))
        except OSError:
            continue
    cands.sort(key=lambda x: x[0], reverse=True)
    now = _t.time()
    out: list[ExternalSession] = []
    for mtime, f in cands[:limit]:
        cwd, title = _droid_session_head(f)
        is_active = bool(cwd) and os.path.realpath(cwd) in owned
        activity = ("turn" if (now - mtime) <= _TURN_RECENT_SECS else "open") if is_active else "idle"
        out.append(ExternalSession(session_id=f.stem, agent_id="droid", cwd=cwd,
                                   last_prompt=title, started_at="", is_active=is_active,
                                   activity=activity))
    return out


class DroidAdapter(PerTurnCLIAdapter):
    cli_name = "droid"
    cap_images = False
    cap_documents = False
    cap_thinking = False         # no thinking event observed; flip True if a reasoning model emits one
    cap_permissions = False      # headless --auto runs unattended; no permission_request blocks
    cap_subagents = True         # Droid exposes a Task tool (subagents)
    cap_interrupt = True         # interrupt = kill the subprocess
    cap_input_during_turn = False
    # Beacon sentinel tools: droid loads the globally-registered beacon MCP from
    # ~/.factory/mcp.json (written by setup_integrations._setup_droid) and expands ${VAR}
    # in the server's env block from `droid exec`'s process env — the same mechanism its
    # BYOK custom models already use for ${OPENROUTER_API_KEY}. `--auto high` auto-approves
    # the MCP tool call (no per-tool prompt). MEASURED end-to-end (P4).
    cap_sentinel = True
    provides_models = True       # parsed from `droid exec --help` (built-in + custom:<id> BYOK)
    or_policy = ORPolicy.BYOK    # shared OR key passed through for custom:<id> models
    provides_usage = True        # completion event carries a usage object
    agent_id = "droid"
    display_name = "Droid"
    cli_binary = "droid"

    def __init__(
        self,
        binary: str | None = None,
        cwd: str | None = None,
        model: str | None = None,
        resume_session_id: str | None = None,
        autonomy: str = "high",
        on_session_id: "Callable[[str], None] | None" = None,
    ) -> None:
        self._binary = binary or shutil.which("droid") or "droid"
        self._cwd = cwd
        self._model = model
        self._session_id: str | None = resume_session_id   # reused via --session-id to keep context
        self._on_session_id = on_session_id   # persists the droid session id for handoff dedup
        self._autonomy = autonomy
        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._tool_meta: dict[str, tuple[str, dict]] = {}  # call id → (name, args) for tool_result
        self._todo_tool_ids: set[str] = set()  # TodoWrite call ids → suppress their tool_result
        self._emitted_text = False   # avoid double-emitting completion.finalText
        self._usage: dict = {}

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

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

    @classmethod
    def transcript_path(cls, ext_sid: str) -> "Path | None":
        """Return the JSONL for `ext_sid` under ~/.factory/sessions/."""
        if not _FACTORY_SESSIONS.exists():
            return None
        for f in _FACTORY_SESSIONS.glob(f"*/{ext_sid}.jsonl"):
            if f.exists():
                return f
        return None

    @classmethod
    def has_changes_since(cls, ts: float) -> bool:
        """True if any droid session file has been written since `ts`."""
        if not _FACTORY_SESSIONS.exists():
            return False
        try:
            if _FACTORY_SESSIONS.stat().st_mtime > ts:
                return True
        except OSError:
            pass
        for f in _FACTORY_SESSIONS.glob("*/*.jsonl"):
            try:
                if f.stat().st_mtime > ts:
                    return True
            except OSError:
                pass
        return False

    @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("/droid"))

    @classmethod
    def latest_sid_for_cwd(cls, cwd: str) -> str | None:
        if not cwd or not _FACTORY_SESSIONS.exists():
            return None
        target = os.path.realpath(os.path.expanduser(cwd))
        newest: tuple[float, str] | None = None
        for f in _FACTORY_SESSIONS.glob("*/*.jsonl"):
            c, _ = _droid_session_head(f)
            if c and os.path.realpath(c) == target:
                try:
                    mt = f.stat().st_mtime
                except OSError:
                    continue
                if newest is None or mt > newest[0]:
                    newest = (mt, f.stem)
        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_droid_mirror
        return make_droid_mirror(sid, from_end=from_end)

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

    @classmethod
    def readiness(cls) -> tuple[bool, str]:
        # A Factory subscription via `droid` login isn't reliably detectable, so we
        # never hard-block. Surface the BYOK path when a shared OpenRouter key exists
        # but no custom model is configured to use it.
        has_custom = False
        try:
            has_custom = bool(json.loads(
                (Path.home() / ".factory" / "settings.json").read_text()).get("customModels"))
        except Exception:
            pass
        if cls.shared_provider_key() and not has_custom:
            return True, ("Add a BYOK custom model (~/.factory/settings.json) to run on your "
                          "OpenRouter key — built-in models need a Factory subscription.")
        return True, ""

    # ── 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="Droid",
            working_dirs=[self._cwd] if self._cwd else [],
        )

    def clear_history(self) -> None:
        # Leaseless: drop the droid session id so the next per-turn spawn omits
        # --session-id (droid mints a fresh one). Wipe the persisted ext_sid too so a
        # reconnect doesn't resume the old session.
        self._session_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]:
        """Parse `droid exec --help` for the model picker.

        The help text carries two two-column sections — `Available Models:`
        (built-in: Claude / GPT / Gemini / Droid Core …) and `Custom Models:`
        (BYOK `custom:<id>` entries from ~/.factory/settings.json). There is no
        machine-readable models command, so the help is the source. Each row is
        `  <id>   <label>`; we split on 2+ spaces. Returns [] on any failure
        (the picker then just shows free-text / the default).
        """
        binary = shutil.which("droid") or "droid"
        try:
            proc = await asyncio.create_subprocess_exec(
                binary, "exec", "--help",
                stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL,
                env=agent_env(),
            )
            out, _ = await asyncio.wait_for(proc.communicate(), timeout=10)
        except Exception:
            return []

        import re as _re
        models: list[dict] = []
        seen: set[str] = set()
        section = False
        for line in out.decode(errors="replace").splitlines():
            header = line.strip().rstrip(":").lower()
            if header in ("available models", "custom models"):
                section = True
                continue
            if section:
                # Section ends at a non-indented line or a blank gap into the next header.
                if line and not line.startswith("  "):
                    section = False
                    continue
                m = _re.match(r"\s{2,}(\S+)\s{2,}(.+?)\s*$", line)
                if not m:
                    continue
                mid, label = m.group(1), m.group(2).strip()
                if mid in seen:
                    continue
                seen.add(mid)
                detail = "BYOK" if mid.startswith("custom:") else ""
                models.append({"id": mid, "label": label or mid, "detail": detail})
        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, "exec",
            "--output-format", "stream-json",
            "--auto", self._autonomy,
        ]
        if self._model:
            cmd += ["--model", self._model]
        if self._session_id:
            cmd += ["--session-id", self._session_id]
        # droid exec has no verified native image/file input → inline attachments as text
        # (Mac-OCR for images + path hints) so they're not silently dropped.
        preamble = self.sentinel_client_preamble(is_first_turn=not self._session_id)
        cmd.append(preamble + await asyncio.to_thread(build_text_with_inline, text, attachments))
        return cmd

    def _pre_spawn(self) -> None:
        # Sentinel: keep ~/.terminull/sentinel_mcp.py current (TOOL_DEFS may have changed since
        # setup). droid reads the server registration from ~/.factory/mcp.json; routing reaches
        # the MCP child via ${TERMINULL_SENTINEL_PORT}/${TERMINULL_SESSION_ID} expanded from spawn env.
        if type(self).cap_sentinel:
            try:
                from ..sentinel_tools import write_global_script
                write_global_script()
            except Exception:
                log.debug("droid: sentinel write_global_script failed", exc_info=True)

    def _turn_env(self) -> dict[str, str]:
        # BYOK: pass the shared OpenRouter key through so `custom:<id>` models route
        # via OpenRouter (default Factory models 402 without a subscription).
        return self.shared_provider_injection(self._model).env

    def _reset_turn_state(self) -> None:
        self._emitted_text = False
        self._turn_ended = False

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

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

        # session_id rides on every event — capture for --session-id resume.
        if sid := event.get("session_id"):
            if sid != self._session_id and self._on_session_id:
                self._on_session_id(sid)
            self._session_id = sid

        if etype == "system" and event.get("subtype") == "init":
            yield self.make_agent_info(
                name="Droid",
                version=str(event.get("model", "")),
                working_dirs=[self._cwd] if self._cwd else [],
            )

        elif etype == "message":
            # role "user" is our own prompt echoed back — skip. Only assistant text.
            if event.get("role") == "assistant":
                txt = event.get("text") or ""
                if txt:
                    self._emitted_text = True
                    yield TextBlock(text=txt, done=True, id=str(uuid.uuid4()))

        elif etype == "tool_call":
            name = event.get("toolName") or event.get("toolId") or "tool"
            params = event.get("parameters")
            args: dict = params if isinstance(params, dict) else {}
            call_id = event.get("id") or str(uuid.uuid4())
            if name.lower() == "todowrite":
                # Factory's plan tool. The whole list is re-sent on every change →
                # emit one live TodoBlock that iOS upserts in place, and suppress
                # the noisy "TODO List Updated" tool_result.
                self._todo_tool_ids.add(call_id)
                yield self.todo_block(self._session_id or "droid",
                                      _parse_droid_todos(args.get("todos")))
                yield AudioHintBlock(state="agent_silent")
                return
            self._tool_meta[call_id] = (name, args)
            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 etype == "tool_result":
            call_id = event.get("id") or str(uuid.uuid4())
            if call_id in self._todo_tool_ids:
                self._todo_tool_ids.discard(call_id)
                return  # suppressed — the live card already reflects this update
            name, args = self._tool_meta.get(call_id, (event.get("toolId") or "tool", {}))
            is_error = bool(event.get("isError"))
            lbl, subj = tool_label_subject(name, args)
            yield ToolCallBlock(id=call_id, name=name, args=args,
                                status="error" if is_error else "done", label=lbl, subject=subj)
            content = _coerce_text(event.get("value"))
            if content:
                yield tool_result_or_large(
                    tool_call_id=call_id, content=content,
                    fetch_store=self._fetch_store, is_error=is_error)
            yield AudioHintBlock(state="agent_speaking")

        elif etype == "completion":
            u = event.get("usage") or {}
            self._usage = {
                "input_tokens": u.get("input_tokens", 0),
                "output_tokens": u.get("output_tokens", 0),
                "cache_read_tokens": u.get("cache_read_input_tokens", 0),
                "cache_creation_tokens": u.get("cache_creation_input_tokens", 0),
            }
            # Fallback: surface finalText if no assistant message streamed this turn.
            final = event.get("finalText")
            if final and not self._emitted_text:
                yield TextBlock(text=str(final), done=True, id=str(uuid.uuid4()))
            yield SessionStatusHintBlock(
                model=self._model or "",
                output_tokens=self._usage["output_tokens"],
                last_input_tokens=self._usage["input_tokens"],
                cache_creation_tokens=self._usage["cache_creation_tokens"],
            )
            yield AudioHintBlock(state="agent_silent")
            if not self._turn_ended:
                self._turn_ended = True
                yield SessionEndBlock(reason="completed")

        elif etype == "error":
            yield AudioHintBlock(state="agent_silent")
            yield ErrorBlock(message=str(event.get("message", "droid error")), recoverable=True)
            # A turn can emit several error events (e.g. agent_loop 402 then a cli
            # "Exec failed"); end the turn exactly once.
            if not self._turn_ended:
                self._turn_ended = True
                yield SessionEndBlock(reason="error")


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

_TODO_LINE = re.compile(r"^\s*(?:\d+[.)]|[-*])?\s*\[(?P<status>[a-zA-Z_ ]+)\]\s*(?P<content>.+?)\s*$")


def _parse_todo_line(line: str) -> dict:
    """Parse one `1. [in_progress] Do the thing` line → {content, status}."""
    m = _TODO_LINE.match(line)
    if m:
        status = m.group("status").strip().lower().replace(" ", "_")
        if status not in ("pending", "in_progress", "completed"):
            status = "completed" if status in ("done", "complete", "x", "✓") else "pending"
        return {"content": m.group("content").strip(), "status": status}
    # No [status] marker → strip a leading "N." / bullet, default pending.
    return {"content": re.sub(r"^\s*(?:\d+[.)]|[-*])\s*", "", line).strip(),
            "status": "pending"}


def _parse_droid_todos(todos) -> list[dict]:
    """Normalise droid's TodoWrite `todos` arg → [{content, status}].

    Factory's TodoWrite is model-dependent in how it serialises the plan: some
    models pass a structured list ([{content/text, status}]), others a single
    formatted string ('1. [pending] Write the script\\n2. [in_progress] ...').
    Handle both so the live card works regardless of the backing model.
    """
    if isinstance(todos, str):
        return [_parse_todo_line(ln) for ln in todos.splitlines() if ln.strip()]
    if isinstance(todos, list):
        out: list[dict] = []
        for t in todos:
            if isinstance(t, dict):
                out.append({
                    "content": t.get("content") or t.get("text") or t.get("task", ""),
                    "status": t.get("status", "pending"),
                })
            elif isinstance(t, str) and t.strip():
                out.append(_parse_todo_line(t))
        return out
    return []


def _coerce_text(value) -> str:
    """tool_result.value is usually a string; coerce structured values to text."""
    if value is None:
        return ""
    if isinstance(value, str):
        return value
    try:
        return json.dumps(value)
    except Exception:
        return str(value)
