"""
GeminiCLIAdapter — drives `gemini` CLI via per-turn subprocess.

Each turn spawns:
  gemini -p <text> --output-format stream-json --skip-trust
         --approval-mode yolo [--resume <session_id>]

Session continuity: Gemini owns the history; we persist the session_id it assigns
on the first run and pass --resume on subsequent turns.

Headless gotchas (verified against gemini-cli 0.46, Jun 2026):
  • --skip-trust is REQUIRED: gemini refuses to run in an "untrusted" folder in
    headless mode (the error goes to stderr only, leaving an empty stdout). Any
    bridge session cwd is untrusted, so without this every turn silently fails.
  • --approval-mode yolo is REQUIRED: the default mode prompts for tool approval,
    which has no answer in non-interactive mode → the turn hangs. We have no
    permission UI for gemini (cap_permissions=False), so we auto-approve.

stream-json event types (Gemini CLI, captured Jun 2026):
  {type: "init",        session_id, model}             ← NO tools list
  {type: "message",     role: "user"|"assistant", content: "..."}
        · role=="user"      → the prompt echoed back; SKIP it
        · role=="assistant" → streamed in many small chunks (deltas), not one blob
  {type: "tool_use",    tool_name, tool_id, parameters: {...}}
  {type: "tool_result", tool_id, status: "success"|"error", output: "..."}
  {type: "error",       message: "..."}
  {type: "result",      status: "success"|"error", stats: {...}}  ← turn complete

Note: gemini has no structured todo/checklist tool. Its "plan" is enter_plan_mode
+ update_topic (a title/summary header) + a markdown plan file written via
write_file — there are no per-step status transitions, so TodoBlock does not apply.
"""
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
from ..blocks import (
    Block, AgentInfoBlock, AudioHintBlock, ErrorBlock, SessionEndBlock,
    SessionStatusHintBlock, SystemStatusBlock, TextBlock, ToolCallBlock,
    tool_result_or_large,
)
from .base import (PerTurnCLIAdapter, ExternalSession, active_cwds_for_command,
                   context_window_for_model, read_stderr_tail)

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

_GEMINI_DIR = Path.home() / ".gemini"
# Gemini's free-tier (OAuth) backend; returns the account's plan/tier but no models.
_CODE_ASSIST_URL = "https://cloudcode-pa.googleapis.com/v1internal:loadCodeAssist"


def _msg_text(msg: dict) -> str:
    c = msg.get("content")
    if isinstance(c, str):
        return c
    if isinstance(c, list):
        return " ".join(p.get("text", "") for p in c if isinstance(p, dict) and p.get("text"))
    return ""


def _parse_gemini_session(f: Path) -> tuple[str, str, str]:
    """(session_id, first_real_user_prompt, started_at) from a chat .jsonl.

    Line 0 is a header; the first message is wrapped in `$set.messages`; later
    messages are appended as bare top-level objects ({id, type, content, …}).
    """
    sid = started = first_user = ""

    def _consider(msg: object) -> None:
        nonlocal first_user
        if first_user or not isinstance(msg, dict) or msg.get("type") != "user":
            return
        txt = _msg_text(msg).strip()
        # The first user turn is always a <session_context> preamble — skip it.
        if txt and not txt.startswith("<session_context>"):
            first_user = txt

    with f.open() as fh:
        for i, line in enumerate(fh):
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except json.JSONDecodeError:
                continue
            if i == 0:
                sid = obj.get("sessionId", "")
                started = obj.get("startTime", "") or obj.get("lastUpdated", "")
                continue
            if "type" in obj and "content" in obj:        # bare appended message
                _consider(obj)
            for key in ("$set", "$push"):                  # batched messages
                blk = obj.get(key)
                if isinstance(blk, dict) and "messages" in blk:
                    ms = blk["messages"]
                    for msg in (ms if isinstance(ms, list) else [ms]):
                        _consider(msg)
            if first_user:
                break
    return sid, first_user, started


def _scan_gemini_sessions(limit: int = 20) -> list[ExternalSession]:
    """Scan ~/.gemini/tmp/*/chats/session-*.jsonl for sessions started via the CLI."""
    tmp = _GEMINI_DIR / "tmp"
    if not tmp.is_dir():
        return []
    # projects.json maps full cwd → project dir name; invert it to recover cwd.
    dir_to_cwd: dict[str, str] = {}
    try:
        projects = json.loads((_GEMINI_DIR / "projects.json").read_text()).get("projects", {})
        dir_to_cwd = {dirname: path for path, dirname in projects.items()}
    except Exception:
        pass

    files: list[Path] = []
    for proj in tmp.iterdir():
        chats = proj / "chats"
        if chats.is_dir():
            files.extend(chats.glob("session-*.jsonl"))
    files.sort(key=lambda p: p.stat().st_mtime, reverse=True)

    result: list[ExternalSession] = []
    for f in files:
        try:
            sid, prompt, started = _parse_gemini_session(f)
        except Exception:
            continue
        if not sid or not prompt:
            continue
        result.append(ExternalSession(
            session_id=sid,
            agent_id="gemini",
            cwd=dir_to_cwd.get(f.parent.parent.name, ""),
            last_prompt=prompt[:200],
            started_at=started,
            is_active=False,
        ))
        if len(result) >= limit:
            break
    return result


class GeminiCLIAdapter(PerTurnCLIAdapter):
    cli_name = "gemini"
    cap_images = True
    cap_documents = True  # CLI supports @<path> file references
    cap_sentinel = False        # P4: beacon MCP not delivered at turn time (mcp.json not loaded in spawn mode)
    cap_interrupt = True
    provides_models = True
    provides_usage = True
    agent_id = "gemini"
    display_name = "Gemini CLI"
    beta = True               # no sentinel (IneligibleTierError upstream) + no structured
                              # todo — the only featured-tier gaps, so kept in the Beta group
    cli_binary = "gemini"

    def __init__(
        self,
        binary: str | None = None,
        cwd: str | None = None,
        auto_approve: bool = False,
        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("gemini") or "gemini"
        self._cwd = cwd
        self._auto_approve = auto_approve
        self._model_pref: str = model or ""   # user-chosen model for `-m`; "" = gemini default
        # Resume an external gemini session (it resumes by id via --resume); None = fresh.
        self._gemini_session_id: str | None = resume_session_id
        self._on_session_id = on_session_id   # persists the gemini session id for handoff dedup
        self._model: str = ""                 # model gemini actually resolved (from init)
        self._turns: int = 0
        self._total_input: int = 0            # accumulated across turns for /status
        self._total_output: int = 0
        self._last_input: int = 0             # last turn's input tokens (≈ context size)
        self._tier_rows: list[dict] = []      # plan/limits rows for /status (fetched once)
        self._send_queue: asyncio.Queue[tuple[str, list[Attachment]]] = None  # type: ignore
        self._tools: list[dict] = []
        self._pending_tools: dict[str, tuple[str, dict]] = {}  # tool_id → (name, args)
        self._fetch_store: dict[str, str] = {}
        self._proc: asyncio.subprocess.Process | None = None
        self._streamed_text: bool = False  # any assistant delta this turn → finalize on result

    # ── Model picker ────────────────────────────────────────────────────────
    # Prefer the live Generative Language models.list (needs GEMINI_API_KEY/
    # GOOGLE_API_KEY; cached per process). The gemini-cli OAuth token can't list
    # models (403), and there's no CLI model-list command, so OAuth-only users get
    # this curated static fallback (newest first). Choice is passed via `-m`.
    @classmethod
    async def list_models(cls, extra: dict | None = None) -> list[dict]:
        from .base import fetch_gemini_models
        return await fetch_gemini_models(fallback=[
            {"id": "gemini-3-pro-preview",   "label": "Gemini 3 Pro",        "detail": "most capable"},
            {"id": "gemini-3-flash-preview", "label": "Gemini 3 Flash",      "detail": "fast (default)"},
            {"id": "gemini-2.5-pro",         "label": "Gemini 2.5 Pro",      "detail": "stable"},
            {"id": "gemini-2.5-flash",       "label": "Gemini 2.5 Flash",    "detail": "stable, fast"},
            {"id": "gemini-2.5-flash-lite",  "label": "Gemini 2.5 Flash-Lite", "detail": "cheapest"},
        ])

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

    @classmethod
    def transcript_path(cls, ext_sid: str) -> "Path | None":
        """Find the gemini chat JSONL whose sessionId header matches `ext_sid`."""
        tmp = _GEMINI_DIR / "tmp"
        if not tmp.is_dir():
            return None
        for f in tmp.glob("*/chats/session-*.jsonl"):
            try:
                sid, _, _ = _parse_gemini_session(f)
                if sid == ext_sid:
                    return f
            except Exception:
                pass
        return None

    @classmethod
    def has_changes_since(cls, ts: float) -> bool:
        """True if any gemini chat file has been written since `ts`."""
        tmp = _GEMINI_DIR / "tmp"
        if not tmp.is_dir():
            return False
        try:
            if tmp.stat().st_mtime > ts:
                return True
        except OSError:
            pass
        for f in tmp.glob("*/chats/session-*.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:
        # Partial handoff: contention guard only (no live mirror). Gemini resumes by id
        # (--resume), so attach restores context; there's no transcript mirror wired yet.
        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("gemini"))

    # ── /status — plan + limits + usage ───────────────────────────────────────
    async def get_usage(self) -> dict:
        return {
            "model": self._model or self._model_pref,
            "turns": self._turns,
            "input_tokens": self._total_input,
            "output_tokens": self._total_output,
            "last_input_tokens": self._last_input,
            "extra_rows": self._tier_rows,   # plan/tier/upgrade rows for the /status card
        }

    async def _fetch_tier(self) -> None:
        """Fetch the account plan/tier from Code Assist (the only limits gemini's
        OAuth backend exposes — there are no numeric quotas in the response).
        Best-effort, cached; surfaced in /status via extra_rows."""
        token = ""
        try:
            token = json.loads((_GEMINI_DIR / "oauth_creds.json").read_text()).get("access_token", "")
        except Exception:
            return
        if not token:
            return
        try:
            import urllib.request
            req = urllib.request.Request(
                _CODE_ASSIST_URL,
                data=json.dumps({"metadata": {"pluginType": "GEMINI"}}).encode(),
                headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
                method="POST",
            )
            data = await asyncio.get_event_loop().run_in_executor(
                None, lambda: json.loads(urllib.request.urlopen(req, timeout=8).read())
            )
        except Exception as e:
            log.debug("gemini _fetch_tier failed: %s", e)
            return
        rows: list[dict] = []
        tier = data.get("currentTier") or {}
        if tier.get("name"):
            rows.append({"label": "Plan", "value": tier["name"]})
        if tier.get("id"):
            rows.append({"label": "Tier", "value": tier["id"]})
        if data.get("upgradeSubscriptionUri"):
            rows.append({"label": "Upgrade", "value": data["upgradeSubscriptionUri"]})
        self._tier_rows = rows

    @classmethod
    def readiness(cls) -> tuple[bool, str]:
        if os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY"):
            return True, ""
        if (Path.home() / ".gemini" / "oauth_creds.json").exists():
            # Runs on OAuth, but the model picker can't list models without an API key.
            return True, "Set GEMINI_API_KEY to list models in the picker."
        return False, "Sign in with `gemini` (or set GEMINI_API_KEY)."

    # ── 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()
        # Fetch the account plan/tier once (best-effort) for /status.
        asyncio.create_task(self._fetch_tier())
        return self.make_agent_info(
            name="Gemini CLI",
            tools=self._tools,
            working_dirs=[self._cwd] if self._cwd else [],
        )

    def clear_history(self) -> None:
        self._gemini_session_id = None
        # Wipe the persisted ext_sid too, else a reconnect resumes the old gemini session.
        if self._on_session_id:
            self._on_session_id("")

    def set_model(self, model: str) -> None:
        # Each turn respawns gemini with `-m`, so a live switch just changes the
        # preference; it takes effect on the next turn (no process restart needed).
        self._model_pref = model or ""

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

    async def _build_argv(self, text: str, attachments: list[Attachment]) -> list[str]:
        # Gemini ingests files (incl. images) via `@<path>` references in the prompt.
        # Attachments already live on disk (Attachment.local_path), so just append
        # them — no temp copy needed.
        prompt = text
        for att in attachments:
            prompt += f"\n@{att.local_path}"

        cmd = [
            self._binary,
            "--output-format", "stream-json",
            "--skip-trust",            # bridge cwd is always "untrusted" → required headless
            "--approval-mode", "yolo",  # no permission UI → must auto-approve or it hangs
        ]
        if self._model_pref:
            cmd += ["-m", self._model_pref]
        # Resume by the UUID gemini assigned on the first turn (verified: --resume
        # accepts a session UUID, not just "latest"/index).
        if self._gemini_session_id:
            cmd += ["--resume", self._gemini_session_id]
        cmd += ["-p", prompt]
        return cmd

    def _reset_turn_state(self) -> None:
        self._pending_tools.clear()
        self._streamed_text = False

    async def _on_no_session_end(self) -> AsyncIterator[Block]:
        # No terminal `result` event. No events *at all* usually means a startup failure
        # (trust/auth/quota) — surface the stderr tail instead of an opaque "error".
        # gemini retries transient quota internally, so quota text on a turn that DID
        # produce events isn't fatal; we only act on the no-event case (_turn_saw_event).
        assert self._proc is not None
        err = await read_stderr_tail(self._proc)
        yield AudioHintBlock(state="agent_silent")
        rc = self._proc.returncode or 0
        if not self._turn_saw_event and (rc != 0 or err):
            low = err.lower()
            if any(k in low for k in ("exhausted your capacity", "quota", "rate limit", "resource_exhausted")):
                yield SystemStatusBlock(event="rate_limited", message=err[-300:] or "Gemini quota exhausted")
            else:
                yield ErrorBlock(message=(err or f"gemini exited with code {rc}")[-800:], recoverable=True)
        yield SessionEndBlock(reason="completed" if rc == 0 else "error")

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

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

        if etype == "init":
            if sid := event.get("session_id"):
                if sid != self._gemini_session_id and self._on_session_id:
                    self._on_session_id(sid)
                self._gemini_session_id = sid
            if model := event.get("model"):
                self._model = model
            # gemini's init carries no tools list; re-emit AgentInfoBlock only to
            # surface the resolved model in the session header.
            yield self.make_agent_info(
                name="Gemini CLI",
                version=self._model,
                tools=self._tools,
                working_dirs=[self._cwd] if self._cwd else [],
            )
            return

        if etype == "message":
            # Gemini echoes the user's own prompt as role=="user"; only the
            # assistant's reply should render. Assistant text arrives as many small
            # chunks (deltas), so stream them and finalize on `result`.
            if event.get("role") != "assistant":
                return
            content = event.get("content", "")
            if content:
                self._streamed_text = True
                yield TextBlock(text=content, done=False)

        elif etype == "tool_use":
            tc_id = event.get("tool_id") or str(uuid.uuid4())
            name = event.get("tool_name", "unknown")
            args = event.get("parameters", {})
            self._pending_tools[tc_id] = (name, args)
            yield AudioHintBlock(state="agent_silent")
            yield ToolCallBlock(id=tc_id, name=name, args=args, status="running")

        elif etype == "tool_result":
            tc_id = event.get("tool_id") or str(uuid.uuid4())
            is_error = event.get("status") == "error"
            name, args = self._pending_tools.pop(tc_id, ("", {}))
            yield ToolCallBlock(
                id=tc_id,
                name=name,
                args=args,
                status="error" if is_error else "done",
            )
            output = event.get("output", "")
            if output:
                yield tool_result_or_large(
                    tool_call_id=tc_id,
                    content=output if isinstance(output, str) else json.dumps(output),
                    fetch_store=self._fetch_store,
                    is_error=is_error,
                )
            yield AudioHintBlock(state="agent_speaking")

        elif etype == "result":
            if sid := event.get("session_id"):
                if sid != self._gemini_session_id and self._on_session_id:
                    self._on_session_id(sid)
                self._gemini_session_id = sid
            # Finalize any streamed assistant text (iOS dedups deltas with the
            # done=True marker, mirroring the codex adapter).
            if self._streamed_text:
                yield TextBlock(text="", done=True)
                self._streamed_text = False
            # Surface token usage in the session header from result.stats, and
            # accumulate totals for /status (get_usage).
            stats = event.get("stats") or {}
            if stats:
                self._turns += 1
                turn_in = stats.get("input_tokens", 0)
                turn_out = stats.get("output_tokens", 0)
                self._total_input += turn_in
                self._total_output += turn_out
                self._last_input = turn_in
                yield SessionStatusHintBlock(
                    model=self._model,
                    turns=self._turns,
                    output_tokens=turn_out,
                    last_input_tokens=turn_in,
                    context_window=context_window_for_model(self._model),
                )
            yield AudioHintBlock(state="agent_silent")
            is_error = event.get("status") == "error"
            yield SessionEndBlock(reason="error" if is_error else "completed")

        elif etype == "error":
            yield ErrorBlock(
                message=event.get("message", "Gemini error"),
                recoverable=True,
            )
            yield AudioHintBlock(state="agent_silent")
            yield SessionEndBlock(reason="error")
