"""
GooseAdapter — drives the open-source Goose agent (Block / Agentic AI Foundation, Apache-2.0) via
its headless `goose run` mode. Goose is general-purpose (not coding-only), so this is Beacon's
"universal / office" agent: any LLM provider, MCP-native, and equipped with office skills/MCP
servers (see docs/office-skills.md) for Word/Excel/PDF/image work — what non-developers need.

Each turn spawns (headless, JSON event stream on stdout):
  goose run --output-format stream-json --quiet --name <session> [--resume] \
    [--with-builtin developer] [--with-extension "<cmd>"]... -t "<text>"

Per-turn / leaseless (like codex/droid/gemini): a fresh `goose run` per message. Context is kept
by NAMING the session (`--name <session>`) and adding `--resume` on every turn after the first;
Goose persists the named session on disk, so a cold bridge restart resumes it too.

stream-json events (one JSON object per line) — confirmed from a live capture (goose 1.38):
  {"type":"message","message":{"role":"assistant"|"user","content":[ <part>, … ]}}
     part kinds:
       {"type":"text","text":"…"}                       ← assistant text, arrives as DELTAS
       {"type":"toolRequest","id","toolCall":{"value":{"name","arguments"}},"_meta":{…}}
       {"type":"toolResponse","id","toolResult":{"status","value":{"content":[{type:text,text}]}}}
       {"type":"thinking"|"reasoning","text":"…"}        ← handled defensively (reasoning models)
  {"type":"complete","total_tokens","input_tokens","output_tokens"}   ← terminal usage event
  {"type":"error", … }                                                 ← surfaced as ErrorBlock

Provider: env-driven — GOOSE_PROVIDER + GOOSE_MODEL + the provider key. Beacon's shared path is
OpenRouter (or_policy PRIMARY): a bridge model id `openrouter/<slug>` → GOOSE_PROVIDER=openrouter,
GOOSE_MODEL=<slug>, OPENROUTER_API_KEY injected via shared_provider_injection. No model ⇒ Goose
uses whatever provider/model the host config already has.

The `developer` builtin (shell + file edit) is enabled by default so the agent can actually read/
write files — essential for office work until the dedicated office MCP servers are wired
(docs/office-skills.md). Disable via the `builtins` ctor arg.
"""
from __future__ import annotations
import asyncio
import json
import logging
import os
import re
import shutil
import uuid
from typing import AsyncIterator, Callable

from ..attachments import Attachment, build_text_with_inline
from ..blocks import (
    Block, AgentInfoBlock, AudioHintBlock, ErrorBlock, SessionEndBlock,
    SessionStatusHintBlock, TextBlock, ThinkingBlock, ToolCallBlock, tool_result_or_large,
)
from .base import (
    PerTurnCLIAdapter, ORPolicy, agent_env, iter_json_lines, read_stderr_tail,
    tool_label_subject, write_session_claude_md,
)

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


def _sanitize_name(sid: str) -> str:
    """Goose `--name` label: keep it filesystem/CLI-safe and bounded."""
    return (re.sub(r"[^A-Za-z0-9_-]", "_", sid) or "beacon")[:64]


_TODO_CHECKBOX = re.compile(r"^\s*(?:[-*+]|\d+[.)])\s*\[(?P<mark>.)\]\s*(?P<content>.+?)\s*$")
_TODO_BARE = re.compile(r"^\s*(?:[-*+]|\d+[.)])\s+(?P<content>.+?)\s*$")


def _parse_goose_todos(content: str) -> list[dict]:
    """Goose's `todo` builtin is a freeform markdown scratchpad — `todo_write`
    stores a raw string like `- [ ] Research\\n- [x] Implement`. Map its checklist
    lines to the generic [{content, status}] shape the TodoBlock helper wants.

    Checkbox marker → status: space/empty = pending; x/X/✓/✔ = completed;
    -/~/·/>/*/./o (and 'wip') = in_progress. Non-checkbox bullet lines are kept as
    pending items; prose/headers (no bullet) are skipped so a plan preamble doesn't
    become a fake step."""
    out: list[dict] = []
    for line in (content or "").splitlines():
        if not line.strip():
            continue
        m = _TODO_CHECKBOX.match(line)
        if m:
            mark = m.group("mark").strip().lower()
            if mark in ("x", "✓", "✔"):
                status = "completed"
            elif mark in ("", " "):
                status = "pending"
            else:                       # -, ~, /, >, *, ., o … any non-empty non-x
                status = "in_progress"
            out.append({"content": m.group("content").strip(), "status": status})
            continue
        b = _TODO_BARE.match(line)
        if b:
            out.append({"content": b.group("content").strip(), "status": "pending"})
    return out


def _coerce_text(value) -> str:
    """toolResult.value.content[*].text (or a raw string/structure) → flat text."""
    if value is None:
        return ""
    if isinstance(value, str):
        return value
    if isinstance(value, dict):
        # {"content":[{"type":"text","text":"…"}], "structuredContent": …}
        parts = value.get("content")
        if isinstance(parts, list):
            txt = "".join(p.get("text", "") for p in parts if isinstance(p, dict))
            if txt:
                return txt
        try:
            return json.dumps(value)
        except Exception:
            return str(value)
    if isinstance(value, list):
        return "".join(p.get("text", "") for p in value if isinstance(p, dict)) or json.dumps(value)
    return str(value)


class GooseAdapter(PerTurnCLIAdapter):
    cli_name = "goose"
    cap_images = False           # `goose run -t` is text-only → inline OCR/path hints (build_text_with_inline)
    cap_documents = False
    cap_thinking = False         # parsed defensively; flip True once a reasoning model's event is confirmed
    cap_permissions = False      # headless run acts unattended (developer builtin auto-runs tools)
    cap_subagents = False
    cap_interrupt = True         # interrupt = kill the subprocess
    cap_input_during_turn = False
    cap_sentinel = True
    provides_models = True       # OpenRouter catalog (shared key)
    or_policy = ORPolicy.PRIMARY # OpenRouter is Goose's shared default provider here
    # Goose also routes the generic OpenAI-compatible custom endpoint (GOOSE_PROVIDER=
    # openai + OPENAI_HOST — see _build_env), so it opts into both registry entries.
    shared_providers = ("openrouter", "custom")
    provides_usage = True        # `complete` event carries token counts
    agent_id = "goose"
    display_name = "Goose"
    beta = False              # feature-complete (model + todo + sentinel, live-verified) — promoted out of Beta
    cli_binary = "goose"

    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,
        builtins: list[str] | None = None,
        extensions: list[str] | None = None,
        isolated_home: bool = False,
        skills_context: dict | None = None,
    ) -> None:
        self._binary = binary or shutil.which("goose") or "goose"
        self._cwd = cwd
        self._model = model
        # Multi-tenant on a shared runner: pin HOME under cwd so Goose's session store
        # (~/.local/share/goose/sessions) is per-cwd, not co-mingled across rooms. Off for the
        # normal bridge (the user's real HOME holds their goose config/auth).
        self._isolated_home = isolated_home
        self._skills_context = skills_context     # {scope: …} baked into remote skill (RAG) URLs
        # Goose has no server-assigned session id — the --name label IS the identity. Reuse the
        # resumed name if given, else derive from the beacon session id in start().
        self._session_name: str | None = resume_session_id or None
        self._resumed = bool(resume_session_id)     # add --resume once a named session exists
        self._on_session_id = on_session_id
        self._builtins = ["developer"] if builtins is None else builtins
        self._extensions = extensions or []         # stdio MCP server commands for --with-extension
        self._proc: asyncio.subprocess.Process | None = None
        self._send_queue: asyncio.Queue[tuple[str, list[Attachment]]] = None  # type: ignore
        self._fetch_store: dict[str, str] = {}
        self._tool_meta: dict[str, tuple[str, dict]] = {}   # call id → (name, args)
        self._todo_tool_ids: set[str] = set()               # todo__* call ids → suppress raw card/result
        self._usage: dict = {}
        self._cur_text = ""                         # accumulates streamed text deltas of one run
        self._cur_text_id = ""
        self._beacon_session_id = ""

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

    async def start(self, session_id: str) -> AgentInfoBlock:
        self._beacon_session_id = session_id
        self._send_queue = asyncio.Queue()
        if not self._session_name:
            self._session_name = _sanitize_name(session_id)
        return self.make_agent_info(
            name="Goose",
            working_dirs=[self._cwd] if self._cwd else [],
        )

    def clear_history(self) -> None:
        # Leaseless: mint a fresh session name so the next spawn starts a NEW Goose session, and
        # wipe the persisted ext id so a reconnect doesn't resume the old one.
        self._session_name = _sanitize_name(f"{self._beacon_session_id}-{uuid.uuid4().hex[:8]}")
        self._resumed = False
        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]:
        """Shared catalog: OpenRouter (GOOSE_PROVIDER=openrouter) + the custom
        OpenAI-compatible endpoint when configured (GOOSE_PROVIDER=openai + OPENAI_HOST)."""
        return await cls.shared_catalog_models()

    # ── Turn loop (send/stream/stop inherited from PerTurnCLIAdapter; goose keeps
    #    its own _run_turn for the --resume→fresh retry + trailing-text flush) ──

    def _resolve_provider_model(self) -> tuple[str | None, str | None, dict[str, str]]:
        """Map the bridge model id → (goose_provider, goose_model, extra_env).

        Single source for both the CLI flags (--provider/--model) and the env, so
        they can't drift. Returns (None, None, env) when no shared provider is
        active (⇒ goose falls back to its own ~/.config/goose/config.yaml)."""
        inj = self.shared_provider_injection(self._model)
        extra: dict[str, str] = dict(inj.env)
        if not (inj.active and inj.provider_id):
            return None, None, extra
        model_arg = self._model
        if model_arg and model_arg.startswith(inj.provider_id + "/"):
            # bridge id `openrouter/openai/gpt-4o-mini` → goose model `openai/gpt-4o-mini`
            model_arg = model_arg[len(inj.provider_id) + 1:]
        if inj.provider_id == "custom":
            # Generic OpenAI-compatible endpoint → Goose's `openai` provider with a
            # custom host. inj.base_url is the /v1 root; Goose splits it into
            # OPENAI_HOST + OPENAI_BASE_PATH, and inj.env already carries
            # OPENAI_API_KEY (the registry env_var for "custom").
            provider = "openai"
            extra["OPENAI_HOST"] = inj.base_url.removesuffix("/v1")
            extra["OPENAI_BASE_PATH"] = "v1/chat/completions"
        else:
            provider = inj.provider_id
        extra["GOOSE_PROVIDER"] = provider
        if model_arg:
            extra["GOOSE_MODEL"] = model_arg
        return provider, model_arg, extra

    def _build_env(self) -> dict[str, str]:
        """GOOSE_PROVIDER/MODEL + the shared provider key, mapped from the bridge model id."""
        _provider, _model, extra = self._resolve_provider_model()
        if self._isolated_home and self._cwd:
            home = os.path.join(self._cwd, ".home")
            os.makedirs(home, exist_ok=True)
            extra["HOME"] = home
        return agent_env(extra, session_id=self._beacon_session_id)

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

        used_resume = self._resumed        # capture: whether THIS run passes --resume
        cmd = [self._binary, "run", "--output-format", "stream-json", "--quiet",
               "--name", self._session_name or "beacon"]
        # Pin provider/model on the CLI, not just via env. `goose run --provider/--model`
        # OVERRIDES both GOOSE_PROVIDER/GOOSE_MODEL env AND the host ~/.config/goose/
        # config.yaml for this run — without it, a bridge that sets no GOOSE_MODEL (no
        # model selected, or a model not namespaced under a provider we hold a key for)
        # silently ran goose's own config default (e.g. active_provider=openrouter,
        # model=anthropic/claude-sonnet-4) instead of what we routed. Flags win, so the
        # model we resolve is the model that runs.
        _provider, _model, _ = self._resolve_provider_model()
        if _provider:
            cmd += ["--provider", _provider]
        if _model:
            cmd += ["--model", _model]
        log.info("goose run provider=%s model=%s (bridge model=%s)",
                 _provider, _model, self._model)
        if used_resume:
            cmd.append("--resume")
        for b in self._builtins:
            cmd += ["--with-builtin", b]
        for ext in self._extensions:
            cmd += ["--with-extension", ext]
        # Terminull sentinel MCP: run_terminal_command / show_photo / schedule + command
        # tools. goose `--with-extension` takes 'ENV=val… command args' and passes that env
        # to the stdio child (like opencode), so the SAME global script every adapter uses is
        # wired here with the port + this session's id. Skipped if the SentinelServer port
        # can't be resolved (headless without sentinel) — goose just runs without the tools.
        import shutil as _shutil
        from ..sentinel_tools import resolve_sentinel_port, write_global_script, _SCRIPT_FILE
        _port = resolve_sentinel_port()
        if _port and self._beacon_session_id:
            write_global_script()   # ensure ~/.terminull/sentinel_mcp.py exists + current
            _py = _shutil.which("python3") or "python3"
            cmd += ["--with-extension",
                    f"TERMINULL_SENTINEL_PORT={_port} "
                    f"TERMINULL_SESSION_ID={self._beacon_session_id} "
                    f"{_py} {_SCRIPT_FILE}"]
        # Office skills: mount the shared, agent-agnostic document MCP servers (Word/Excel/PDF/
        # images) so this universal/office agent can read+edit real files. Same servers every
        # adapter uses — see office_skills.py + docs/office-skills.md.
        from ..office_skills import goose_extension_flags
        cmd += goose_extension_flags(context=self._skills_context)
        preamble = self.sentinel_client_preamble(is_first_turn=not used_resume)
        cmd += ["-t", preamble + await asyncio.to_thread(build_text_with_inline, text, attachments)]

        write_session_claude_md(self._cwd)
        try:
            self._proc = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                cwd=self._cwd,
                env=self._build_env(),
            )
        except Exception as e:
            yield ErrorBlock(message=f"Failed to start goose: {e}", recoverable=True)
            yield SessionEndBlock(reason="error")
            return

        # A named session now exists on disk → resume it from the next turn on.
        self._resumed = True
        if self._on_session_id and self._session_name:
            self._on_session_id(self._session_name)

        self._cur_text, self._cur_text_id = "", ""
        session_ended = False
        try:
            assert self._proc.stdout is not None
            async for event in iter_json_lines(self._proc.stdout):
                async for block in self._parse_event(event):
                    if isinstance(block, SessionEndBlock):
                        session_ended = True
                    yield block
        except Exception as e:
            yield ErrorBlock(message=str(e), recoverable=True)
        finally:
            await self._proc.wait()

        # Flush a trailing text run that never got a terminal `complete`.
        if self._cur_text:
            yield TextBlock(text=self._cur_text, done=True, id=self._cur_text_id)
            self._cur_text = ""

        if not session_ended:
            err_txt = await read_stderr_tail(self._proc)
            # First turn for this session name: `--resume` failed because Goose has no session yet
            # (the gateway always supplies a resume id). Retry once as a FRESH session (no --resume)
            # — nothing was emitted, so this is a clean restart.
            if used_resume and "No session found" in err_txt and not self._cur_text:
                self._resumed = False
                async for blk in self._run_turn(text, attachments):
                    yield blk
                return
            yield AudioHintBlock(state="agent_silent")
            if (self._proc.returncode or 0) != 0 or err_txt:
                yield ErrorBlock(message=err_txt or "goose exited with an error", recoverable=True)
                yield SessionEndBlock(reason="error")
            else:
                yield SessionEndBlock(reason="completed")

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

    def _flush_text(self) -> Block | None:
        """Close the current streamed text run as a final TextBlock (if any)."""
        if self._cur_text:
            blk = TextBlock(text=self._cur_text, done=True, id=self._cur_text_id)
            self._cur_text, self._cur_text_id = "", ""
            return blk
        return None

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

        if etype == "message":
            msg = event.get("message") or {}
            role = msg.get("role")
            for part in (msg.get("content") or []):
                if not isinstance(part, dict):
                    continue
                ptype = part.get("type")

                if ptype == "text" and role == "assistant":
                    if not self._cur_text_id:
                        self._cur_text_id = str(uuid.uuid4())
                    # goose `message` text parts arrive as pure deltas. iOS APPENDS
                    # done=False partials (the universal delta-stream contract — same
                    # as claude_code/hermes), so yield only the NEW chunk, never the
                    # accumulated buffer (that duplicated it per event: "The" →
                    # "TheThe quick" → …). _cur_text stays the full text for the
                    # done=True flush, which iOS replaces the block with.
                    chunk = part.get("text", "")
                    self._cur_text += chunk
                    yield TextBlock(text=chunk, done=False, id=self._cur_text_id)

                elif ptype in ("thinking", "reasoning") and role == "assistant":
                    flushed = self._flush_text()
                    if flushed:
                        yield flushed
                    think = part.get("text") or part.get("thinking") or ""
                    if think:
                        yield ThinkingBlock(thinking=think, done=True)

                elif ptype == "toolRequest":
                    flushed = self._flush_text()
                    if flushed:
                        yield flushed
                    call = (part.get("toolCall") or {}).get("value") or {}
                    name = call.get("name") or "tool"
                    raw_args = call.get("arguments")
                    args: dict = raw_args if isinstance(raw_args, dict) else {}
                    call_id = part.get("id") or str(uuid.uuid4())
                    self._tool_meta[call_id] = (name, args)
                    # goose's `todo` builtin (todo__todo_write / todo__todo_read) is a
                    # markdown checklist scratchpad — surface it as the generic live
                    # TodoBlock plan card (one upsert per session) instead of a raw tool
                    # card, matching claude_code/opencode/kilo. Suppress the tool card +
                    # its "Updated (N chars)" result. `todo_write` carries the new list in
                    # args.content; `todo_read` only reads (its list rides the response).
                    if name in ("todo__todo_write", "todo__todo_read"):
                        self._todo_tool_ids.add(call_id)
                        todos = _parse_goose_todos(args.get("content", ""))
                        if todos:
                            yield self.todo_block(self._beacon_session_id, todos)
                        yield AudioHintBlock(state="agent_silent")
                        continue
                    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 ptype == "toolResponse":
                    res = part.get("toolResult") or {}
                    is_error = res.get("status") not in (None, "success")
                    call_id = part.get("id") or str(uuid.uuid4())
                    # Suppressed todo tool: re-derive the plan from the read/write result
                    # (todo_read returns the current list here) then swallow the raw card.
                    if call_id in self._todo_tool_ids:
                        self._todo_tool_ids.discard(call_id)
                        todos = _parse_goose_todos(_coerce_text(res.get("value")))
                        if todos:
                            yield self.todo_block(self._beacon_session_id, todos)
                        yield AudioHintBlock(state="agent_speaking")
                        continue
                    name, args = self._tool_meta.get(call_id, ("tool", {}))
                    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(res.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 == "complete":
            flushed = self._flush_text()
            if flushed:
                yield flushed
            self._usage = {
                "input_tokens": event.get("input_tokens", 0),
                "output_tokens": event.get("output_tokens", 0),
                "cache_read_tokens": 0,
                "cache_creation_tokens": 0,
            }
            yield SessionStatusHintBlock(
                model=self._model or "",
                output_tokens=self._usage["output_tokens"],
                last_input_tokens=self._usage["input_tokens"],
            )
            yield AudioHintBlock(state="agent_silent")
            yield SessionEndBlock(reason="completed")

        elif etype == "error":
            flushed = self._flush_text()
            if flushed:
                yield flushed
            yield AudioHintBlock(state="agent_silent")
            yield ErrorBlock(message=str(event.get("message") or event.get("error") or "goose error"),
                             recoverable=True)
            yield SessionEndBlock(reason="error")
