"""
OpenCodeAdapter — drives sst/opencode via its headless HTTP server + SSE.

opencode ≥1.x removed the Claude-compatible stream-json stdio mode; it is now a
server (`opencode serve`). This adapter mirrors the kilo adapter (kilo is an
opencode fork, same protocol):

  1. spawn `opencode serve --port 0`  (one server per adapter/cwd)
  2. subscribe to GET /global/event   (SSE; each event is {payload:{type, properties}})
  3. create a session (POST /session) and send prompts (POST /session/{id}/prompt_async)
  4. translate the event stream into Beacon blocks

Event vocabulary (captured from opencode 1.x):
  message.part.delta   {partID, field:"text"|"reasoning", delta}  ← streaming text/thinking
  message.part.updated {part:{type:"text"|"reasoning"|"tool"|"step-*", …}}  ← finalized parts
  todo.updated         {todos:[{content,status}]}                 ← live plan → TodoBlock
  session.error        {error}                                    ← ErrorBlock
  session.idle         {sessionID}                                ← turn complete

Auth: opencode uses its own configured provider credentials. **For Beacon dev
only**, set TERMINULL_DEV_OPENROUTER=1 to run opencode off the shared Beacon
OpenRouter key (base.openrouter_key()) — production users keep their own opencode
login and we never force OpenRouter on them.

Install: npm install -g opencode-ai   (or: brew install sst/tap/opencode)
"""
from __future__ import annotations
import asyncio
import json
import logging
import os
import re
import shutil
import uuid
import sqlite3
from pathlib import Path
from typing import AsyncIterator, Callable

import httpx

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

from ..attachments import Attachment, build_text_with_inline
from ..blocks import (
    Block, AgentInfoBlock, AudioHintBlock, ErrorBlock, SessionEndBlock,
    TextBlock, ThinkingBlock, ToolCallBlock, tool_result_or_large,
)
from .base import (BaseAdapter, ExternalSession, ORPolicy, active_cwds_for_command,
                   agent_env, fetch_openrouter_models, split_markdown, tool_label_subject)

# Dev-mode fallback: OR model used when TERMINULL_DEV_OPENROUTER=1 and no model is
# explicitly selected. Only reachable in dev — OPT_IN policy blocks OR routing in
# production unless the user picks an openrouter/* model from the picker.
_DEFAULT_OR_MODEL = "anthropic/claude-sonnet-4-6"

# Sentinel picker id → use opencode's own configured provider (no OR routing).
_DEFAULT_MODEL_ID = "default"


_OPENCODE_DB = Path.home() / ".local" / "share" / "opencode" / "opencode.db"
# A session in cwd written within this window counts as actively driven elsewhere.
_OC_ACTIVE_WINDOW_MS = 60_000
_OC_NEW_SESSION_RE = re.compile(r"^(New session|Untitled)\s*[-–]?", re.IGNORECASE)


def _oc_connect():
    con = sqlite3.connect(f"file:{_OPENCODE_DB}?mode=ro", uri=True, check_same_thread=False)
    con.row_factory = sqlite3.Row
    return con


def _oc_sessions_in_cwd(cwd: str) -> list[tuple[str, int]]:
    """[(session_id, time_updated_ms)] for non-archived opencode sessions in `cwd`, newest first."""
    if not _OPENCODE_DB.exists() or not cwd:
        return []
    target = os.path.realpath(os.path.expanduser(cwd))
    try:
        con = _oc_connect()
    except Exception:
        return []
    try:
        rows = con.execute("""
            SELECT id, directory, time_updated FROM session
            WHERE time_archived IS NULL AND directory IS NOT NULL
            ORDER BY time_updated DESC
        """).fetchall()
    except Exception:
        return []
    finally:
        con.close()
    out: list[tuple[str, int]] = []
    for r in rows:
        try:
            if os.path.realpath(r["directory"]) == target:
                out.append((r["id"], int(r["time_updated"] or 0)))
        except Exception:
            continue
    return out


def _scan_opencode_sessions(limit: int = 20) -> list[ExternalSession]:
    """Discover opencode sessions from opencode.db (same schema family as kilo.db)."""
    if not _OPENCODE_DB.exists():
        return []
    import time as _t
    running = bool(active_cwds_for_command("/opencode"))
    try:
        con = _oc_connect()
    except Exception:
        return []
    try:
        rows = con.execute("""
            SELECT s.id, s.directory, s.title, datetime(s.time_created/1000,'unixepoch') AS started,
                   s.time_updated,
                   (SELECT json_extract(p.data,'$.text') FROM part p JOIN message m ON p.message_id=m.id
                    WHERE m.session_id=s.id AND json_extract(m.data,'$.role')='user'
                      AND json_extract(p.data,'$.type')='text'
                    ORDER BY m.time_created, p.time_created LIMIT 1) AS first_prompt
            FROM session s
            WHERE s.time_archived IS NULL
              AND EXISTS (SELECT 1 FROM message m WHERE m.session_id=s.id
                          AND json_extract(m.data,'$.role')='user')
            ORDER BY s.time_updated DESC LIMIT ?
        """, (limit,)).fetchall()
    except Exception:
        return []
    finally:
        con.close()
    now_ms = _t.time() * 1000
    out: list[ExternalSession] = []
    for r in rows:
        title = r["title"] or ""
        last_prompt = title if (title and not _OC_NEW_SESSION_RE.match(title)) else (r["first_prompt"] or "")
        started = (r["started"] or "")
        if started:
            started = started.replace(" ", "T") + "Z"
        recent = running and (now_ms - int(r["time_updated"] or 0)) < _OC_ACTIVE_WINDOW_MS
        out.append(ExternalSession(
            session_id=r["id"], agent_id="opencode", cwd=r["directory"] or "",
            last_prompt=last_prompt, started_at=started, is_active=recent,
            activity="open" if recent else "idle"))
    return out


class OpenCodeAdapter(BaseAdapter):
    cap_images = False
    cap_documents = False
    # P4 (measured, stable): the beacon MCP IS delivered at turn time. opencode loads the
    # globally-registered beacon server (the shared stdio script) and its MCP child inherits
    # TERMINULL_SESSION_ID from the server's env (set in _spawn_server) → calls route correctly.
    cap_sentinel = True
    cap_thinking = True       # opencode streams reasoning deltas
    cap_interrupt = True
    # cap_permissions stays False — serve auto-runs tools (no permission gate)
    provides_models = True        # picker = "Default" + shared OpenRouter models (when a key exists)
    provides_usage = False
    or_policy = ORPolicy.OPT_IN   # uses its own provider login unless an OR model is chosen
    agent_id = "opencode"
    display_name = "OpenCode"
    beta = False              # live-verified via the adapter matrix
    cli_binary = "opencode"

    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("opencode") or "opencode"
        self._cwd = cwd
        self._model_pref = model or ""
        self._oc_session_id = resume_session_id or ""
        self._on_session_id = on_session_id   # persists the opencode session id for handoff dedup
        self._beacon_session_id = ""           # set in start(); routes the server's beacon MCP child
        self._base_url = ""
        self._proc: asyncio.subprocess.Process | None = None
        self._send_queue: asyncio.Queue[str] = None  # type: ignore
        self._fetch_store: dict[str, str] = {}
        self._stopping = False
        # per-turn state
        self._part_types: dict[str, str] = {}            # partID → type
        self._text_parts: dict[str, tuple[str, str]] = {}  # partID → (block_id, accumulated)
        self._thinking_parts: dict[str, str] = {}        # partID → accumulated reasoning
        self._pending_tools: dict[str, str] = {}         # callID → block_id

    # ── Session handoff (docs/session-handoff.md) — opencode shares kilo's DB schema ──

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

    @classmethod
    def read_session_condensed(cls, ext_sid: str, max_turns: int = 20) -> str:
        """Return last `max_turns` user+assistant exchanges from opencode.db as condensed text."""
        if not _OPENCODE_DB.exists():
            return ""
        try:
            con = _oc_connect()
            rows = con.execute("""
                SELECT json_extract(m.data, '$.role') AS role,
                       json_extract(p.data, '$.text') AS text
                FROM   part p
                JOIN   message m ON p.message_id = m.id
                WHERE  m.session_id = ?
                  AND  json_extract(p.data, '$.type') = 'text'
                  AND  json_extract(m.data, '$.role') IN ('user', 'assistant')
                ORDER BY m.time_created, p.time_created
            """, (ext_sid,)).fetchall()
            con.close()
        except Exception:
            return ""
        turns = [f"{r['role']}: {(r['text'] or '')[:400]}" for r in rows if r['text']]
        return "\n\n".join(turns[-(max_turns * 2):])

    @classmethod
    def has_changes_since(cls, ts: float) -> bool:
        """True if opencode.db has been modified since `ts`."""
        try:
            return _OPENCODE_DB.exists() and _OPENCODE_DB.stat().st_mtime > ts
        except OSError:
            return False

    @classmethod
    def is_cwd_externally_owned(cls, cwd: str, sid: str = "") -> bool:
        if not cwd or not active_cwds_for_command("/opencode"):
            return False
        import time as _t
        now_ms = _t.time() * 1000
        return any((now_ms - ts) < _OC_ACTIVE_WINDOW_MS for _id, ts in _oc_sessions_in_cwd(cwd))

    @classmethod
    def latest_sid_for_cwd(cls, cwd: str) -> str | None:
        s = _oc_sessions_in_cwd(cwd)
        return s[0][0] if s else None

    @classmethod
    def make_transcript_mirror(cls, cwd: str, sid: str, *, from_end: bool = False):
        # opencode.db has the same session/message/part schema as kilo.db.
        from ..transcript_mirror import KiloTranscriptMirror
        return KiloTranscriptMirror(_OPENCODE_DB, sid, from_end=from_end) if (sid and _OPENCODE_DB.exists()) else None

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

    @classmethod
    def readiness(cls) -> tuple[bool, str]:
        home = Path.home()
        if any((home / p).exists() for p in
               (".local/share/opencode/auth.json", ".config/opencode/auth.json")):
            return True, ""
        if cls.shared_provider_key():
            return True, "No OpenCode login — pick an OpenRouter model to use the shared key."
        return False, "Run `opencode auth login` (or add an OpenRouter key)."

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

    async def start(self, session_id: str) -> AgentInfoBlock:
        self._beacon_session_id = session_id   # routes the server's beacon MCP child
        self._send_queue = asyncio.Queue()
        await self._spawn_server()
        if not self._oc_session_id:
            self._oc_session_id = await self._create_session()
            if self._oc_session_id and self._on_session_id:
                self._on_session_id(self._oc_session_id)
        return self.make_agent_info(
            name="OpenCode",
            working_dirs=[self._cwd] if self._cwd else [],
        )

    async def _spawn_server(self) -> None:
        # Lend the shared OpenRouter key to the server whenever one is discoverable
        # and the bridge gate allows it — regardless of the current model — so that
        # selecting an openrouter/* model mid-session works without a respawn. The
        # key is inert unless a turn's payload routes providerID=openrouter (OPT_IN
        # consent in _model_payload); a personal bridge can disable lending via the
        # [providers] share_openrouter_key gate.
        pk = self.shared_provider_key()
        extra: dict[str, str] = {"OPENROUTER_API_KEY": pk.key} if pk else {}

        cmd = [self._binary, "serve", "--port", "0", "--hostname", "127.0.0.1"]
        env = agent_env(extra, session_id=self._beacon_session_id)
        # Ensure the spawned agent has access to the full user PATH to prevent "restricted environment"
        # flags (like the "try disabling bash" hint) in environments where the Bridge subprocess
        # might otherwise inherit a narrow path.
        env["PATH"] = os.environ.get("PATH", env.get("PATH", ""))

        self._proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.STDOUT,
            cwd=self._cwd,
            env=env,
        )
        port: int | None = None
        assert self._proc.stdout is not None
        for _ in range(40):
            raw = await asyncio.wait_for(self._proc.stdout.readline(), timeout=20)
            if not raw:
                break
            m = re.search(r"(?:127\.0\.0\.1|localhost):(\d+)", raw.decode(errors="replace"))
            if m:
                port = int(m.group(1))
                break
        if not port:
            raise RuntimeError("opencode serve did not print a port within timeout")
        self._base_url = f"http://127.0.0.1:{port}"

    async def _create_session(self) -> str:
        async with httpx.AsyncClient() as client:
            resp = await client.post(f"{self._base_url}/session", json={}, timeout=10)
            resp.raise_for_status()
            return resp.json()["id"]

    def clear_history(self) -> None:
        self._oc_session_id = ""
        # Wipe the persisted ext_sid too, else a reconnect resumes the old opencode
        # session. Next send creates a fresh server session (see _create_session).
        if self._on_session_id:
            self._on_session_id("")

    def set_model(self, model: str) -> None:
        # prompt model is sent per turn, so a switch just changes the preference.
        self._model_pref = model or ""

    async def stop(self) -> None:
        self._stopping = True
        if self._base_url and self._oc_session_id:
            try:
                async with httpx.AsyncClient() as client:
                    await client.post(f"{self._base_url}/session/{self._oc_session_id}/abort", timeout=5)
            except Exception:
                pass
        if self._proc and self._proc.returncode is None:
            self._proc.terminate()
            try:
                await asyncio.wait_for(self._proc.wait(), timeout=5)
            except asyncio.TimeoutError:
                self._proc.kill()

    # ── Model payload ──────────────────────────────────────────────────────────

    @classmethod
    async def list_models(cls, extra: dict | None = None) -> list[dict]:
        """Picker = "Default (your OpenCode login)" + live OpenRouter model list.

        Only shown when an OR key is discoverable + the bridge gate allows lending
        it. With no key, return [] so opencode shows no picker and runs on its own
        provider auth. Selecting "Default" → own login; selecting an openrouter/*
        model → routed via the shared key (OPT_IN policy)."""
        pk = cls.shared_provider_key()
        if not pk:
            return []
        models: list[dict] = [{"id": _DEFAULT_MODEL_ID, "label": "Default (OpenCode login)",
                                "detail": "your configured provider"}]
        or_models = await fetch_openrouter_models(pk.key)
        for m in or_models:
            models.append({"id": f"openrouter/{m['id']}", "label": m["label"],
                           "detail": "via Beacon OpenRouter"})
        return models

    def _model_payload(self) -> dict | None:
        """{"providerID","modelID"} for the prompt, or None to use opencode's default."""
        if self._model_pref == _DEFAULT_MODEL_ID:
            return None                          # explicit "Default" → own provider login
        inj = self.shared_provider_injection(self._model_pref or None)
        pid = inj.provider_id
        if inj.active and pid:
            mid = (self._model_pref or _DEFAULT_OR_MODEL).removeprefix(pid + "/")
            return {"providerID": pid, "modelID": mid}
        if self._model_pref and "/" in self._model_pref:
            provider, rest = self._model_pref.split("/", 1)
            return {"providerID": provider, "modelID": rest}
        return None

    # ── I/O ───────────────────────────────────────────────────────────────────

    async def send(self, text: str, attachments: list[Attachment] | None = None) -> None:
        # opencode's prompt endpoint takes text parts only (no native image input wired) →
        # inline attachments as text (Mac-OCR for images + path hints). See attachments.py.
        first = not self._sentinel_preamble_sent
        self._sentinel_preamble_sent = True
        await self._send_queue.put(
            self.sentinel_client_preamble(is_first_turn=first)
            + build_text_with_inline(text, attachments or []))

    async def stream(self) -> AsyncIterator[Block]:
        while True:
            text = await self._send_queue.get()
            async for block in self._run_turn(text):
                yield block

    async def _run_turn(self, text: str) -> AsyncIterator[Block]:
        yield AudioHintBlock(state="agent_speaking")

        # Reset per-turn state, then open SSE BEFORE posting to avoid missing events.
        self._stopping = False
        self._part_types.clear()
        self._text_parts.clear()
        self._thinking_parts.clear()
        self._pending_tools.clear()
        event_iter = self._sse_events()

        body: dict = {"parts": [{"type": "text", "text": text}]}
        if (model := self._model_payload()):
            body["model"] = model
        try:
            async with httpx.AsyncClient() as client:
                resp = await client.post(
                    f"{self._base_url}/session/{self._oc_session_id}/prompt_async",
                    json=body, timeout=15,
                )
                if resp.status_code not in (200, 201, 202, 204):
                    yield ErrorBlock(message=f"opencode /prompt_async HTTP {resp.status_code}: {resp.text[:200]}",
                                     recoverable=True)
                    yield SessionEndBlock(reason="error")
                    return
        except Exception as e:
            log.warning("opencode connection failed: %s", e, exc_info=True)
            yield ErrorBlock(message=f"opencode connection failed: {e}", recoverable=True)
            yield SessionEndBlock(reason="error")
            return

        try:
            async for etype, props in event_iter:
                sid = props.get("sessionID") or props.get("part", {}).get("sessionID", "")
                if sid and sid != self._oc_session_id:
                    continue
                done = False
                async for block in self._parse_event(etype, props):
                    if isinstance(block, SessionEndBlock):
                        done = True
                    yield block
                if done:
                    break
        except Exception as e:
            log.warning("opencode event stream failed: %s", e, exc_info=True)
            yield ErrorBlock(message=str(e), recoverable=True)
            yield SessionEndBlock(reason="error")

    async def _sse_events(self) -> AsyncIterator[tuple[str, dict]]:
        """Yield (type, properties) from GET /global/event. Each SSE datum is
        {payload:{type, properties}} — unwrap the payload."""
        async with httpx.AsyncClient(timeout=httpx.Timeout(300, connect=10)) as client:
            async with client.stream(
                "GET", f"{self._base_url}/global/event",
                headers={"Accept": "text/event-stream"},
            ) as response:
                async for line in response.aiter_lines():
                    if not line.startswith("data:"):
                        continue
                    raw = line[5:].strip()
                    if not raw:
                        continue
                    try:
                        event = json.loads(raw)
                    except Exception:
                        continue
                    payload = event.get("payload", event)
                    yield payload.get("type", ""), payload.get("properties", {})

    # ── Tool helpers ──────────────────────────────────────────────────────────

    def _tool_running(self, tool_name: str, call_id: str, args: dict) -> list[Block]:
        block_id = self._pending_tools.get(call_id) or str(uuid.uuid4())
        self._pending_tools[call_id] = block_id
        lbl, subj = tool_label_subject(tool_name, args if isinstance(args, dict) else {})
        return [AudioHintBlock(state="agent_silent"),
                ToolCallBlock(id=block_id, name=tool_name, args=args or {},
                              status="running", label=lbl, subject=subj)]

    def _tool_done(self, tool_name: str, call_id: str, args: dict,
                   result_text: str, is_error: bool) -> list[Block]:
        block_id = self._pending_tools.pop(call_id, str(uuid.uuid4()))
        lbl, subj = tool_label_subject(tool_name, args if isinstance(args, dict) else {})
        return [
            ToolCallBlock(id=block_id, name=tool_name, args=args or {},
                          status="error" if is_error else "done", label=lbl, subject=subj),
            tool_result_or_large(tool_call_id=call_id, content=result_text,
                                 fetch_store=self._fetch_store, is_error=is_error),
            AudioHintBlock(state="agent_speaking"),
        ]

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

    async def _parse_event(self, etype: str, props: dict) -> AsyncIterator[Block]:
        if etype in ("", "server.connected", "server.heartbeat"):
            return

        # ── streaming text / reasoning deltas ────────────────────────────────
        if etype == "message.part.delta":
            part_id = props.get("partID", "")
            if props.get("field") != "text":
                return
            delta = props.get("delta", "")
            # iOS APPENDS done=False partials (the universal delta-stream contract),
            # so yield only the NEW delta, never the accumulated buffer — the buffer
            # duplicated text per event and was only corrected by the done=True
            # snapshot in message.part.updated. We keep accumulating internally so
            # the final done=True block still carries the full text (iOS replaces).
            if self._part_types.get(part_id) == "reasoning":
                self._thinking_parts[part_id] = self._thinking_parts.get(part_id, "") + delta
                yield ThinkingBlock(thinking=delta, done=False)
            else:
                block_id, acc = self._text_parts.get(part_id, (str(uuid.uuid4()), ""))
                acc += delta
                self._text_parts[part_id] = (block_id, acc)
                yield TextBlock(text=delta, done=False, id=block_id)

        # ── finalized parts: text / reasoning / tool ─────────────────────────
        elif etype == "message.part.updated":
            part = props.get("part", {})
            part_id = part.get("id", "")
            part_type = part.get("type", "")
            if part_id and part_type:
                self._part_types[part_id] = part_type

            if part_type == "text":
                text = part.get("text", "")
                if part_id in self._text_parts:
                    block_id, _ = self._text_parts.pop(part_id)
                    yield TextBlock(text=text, done=True, id=block_id)
                    for block in split_markdown(text):
                        if not isinstance(block, TextBlock):
                            yield block
                elif text:
                    for block in split_markdown(text):
                        yield block

            elif part_type == "reasoning":
                text = part.get("text", self._thinking_parts.pop(part_id, ""))
                if text:
                    yield ThinkingBlock(thinking=text, done=True)

            elif part_type == "tool":
                tool_name = part.get("tool", "")
                call_id = part.get("callID", "")
                state = part.get("state", {})
                status = state.get("status", "")
                args = state.get("input") or {}
                if status == "running":
                    for b in self._tool_running(tool_name, call_id, args):
                        yield b
                elif status == "completed":
                    output = state.get("output", "")
                    is_error = isinstance(output, dict) and bool(output.get("isError"))
                    result_text = output if isinstance(output, str) else json.dumps(output)
                    for b in self._tool_done(tool_name, call_id, args, result_text, is_error):
                        yield b
                elif status == "error":
                    for b in self._tool_done(tool_name, call_id, args,
                                             str(state.get("error", "")), True):
                        yield b

        # ── live plan → generic TodoBlock ────────────────────────────────────
        elif etype == "todo.updated":
            todos = props.get("todos", [])
            if todos:
                yield self.todo_block(self._oc_session_id, todos)

        # ── errors + turn end ────────────────────────────────────────────────
        elif etype == "session.error":
            err = props.get("error", {})
            msg = err.get("message") or err.get("data", {}).get("message") or str(err) if isinstance(err, dict) else str(err)
            yield ErrorBlock(message=msg or "opencode error", recoverable=True)
            yield AudioHintBlock(state="agent_silent")
            yield SessionEndBlock(reason="error")

        elif etype == "session.idle":
            # Finalize any unclosed streamed text, then end the turn.
            for part_id, (block_id, acc) in list(self._text_parts.items()):
                yield TextBlock(text=acc, done=True, id=block_id)
            self._text_parts.clear()
            yield AudioHintBlock(state="agent_silent")
            yield SessionEndBlock(reason="completed")
