from __future__ import annotations
import asyncio
import inspect
import json
import logging
import os
import re
import shutil
import subprocess
import urllib.request
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, AsyncIterator, Awaitable, Callable
from ..blocks import Block, AgentInfoBlock, TextBlock, CodeBlock, TodoBlock

log = logging.getLogger(__name__)

_ANTHROPIC_MODELS_URL = "https://api.anthropic.com/v1/models"
_ANTHROPIC_API_VERSION = "2023-06-01"
_ANTHROPIC_OAUTH_USAGE_URL = "https://api.anthropic.com/api/oauth/usage"
_ANTHROPIC_OAUTH_BETA = "oauth-2025-04-20"

# Cached result so keychain is read at most once per process lifetime.
_models_cache: list[dict] | None = None

# Short-lived cache for the OAuth subscription usage (5h / weekly windows).
# Live-ish data, so a small TTL instead of the process-lifetime model cache.
_oauth_usage_cache: tuple[float, dict | None] | None = None
_OAUTH_USAGE_TTL = 60.0  # seconds


def _keychain_oauth_token() -> str:
    """Read the Claude Code OAuth access token from the macOS keychain. Returns '' on failure."""
    try:
        raw = subprocess.check_output(
            ["security", "find-generic-password", "-s", "Claude Code-credentials", "-w"],
            stderr=subprocess.DEVNULL,
            timeout=5,
        ).decode().strip()
        return json.loads(raw).get("claudeAiOauth", {}).get("accessToken", "")
    except Exception:
        return ""


async def fetch_anthropic_models(fallback: list[dict]) -> list[dict]:
    """Fetch the live Anthropic model list.

    Auth priority:
      1. macOS keychain OAuth token (Claude Pro/subscription users)
      2. ANTHROPIC_API_KEY env var (API key users, Linux VPS)
      3. Returns `fallback` if both fail.

    Result is cached for the process lifetime.
    """
    global _models_cache
    if _models_cache is not None:
        return _models_cache

    # _keychain_oauth_token spawns the macOS `security` CLI (sync, up to 5s if the
    # keychain is locked) — keep it off the event loop.
    import asyncio
    token = await asyncio.get_event_loop().run_in_executor(None, _keychain_oauth_token)
    if token:
        headers = {
            "Authorization": f"Bearer {token}",
            "anthropic-version": _ANTHROPIC_API_VERSION,
        }
    else:
        api_key = os.environ.get("ANTHROPIC_API_KEY", "")
        if not api_key:
            log.debug("fetch_anthropic_models: no auth available, using fallback list")
            return fallback
        headers = {
            "x-api-key": api_key,
            "anthropic-version": _ANTHROPIC_API_VERSION,
        }

    try:
        import asyncio, re as _re
        req = urllib.request.Request(_ANTHROPIC_MODELS_URL, headers=headers)
        data = await asyncio.get_event_loop().run_in_executor(
            None, lambda: json.loads(urllib.request.urlopen(req, timeout=10).read())
        )
        all_models = data.get("data", [])

        # Keep only the newest model per family (fable/opus/sonnet/haiku/…).
        # Family = model ID stripped of trailing version numbers and date suffixes.
        _date_re = _re.compile(r"-\d{8}$")
        _ver_re  = _re.compile(r"(-\d+)+$")

        def _family(model_id: str) -> str:
            s = _date_re.sub("", model_id)   # strip -YYYYMMDD
            s = _ver_re.sub("", s)           # strip trailing -X[-Y…]
            return s                         # e.g. "claude-opus", "claude-sonnet"

        seen: set[str] = set()
        filtered: list[dict] = []
        for m in all_models:                 # already sorted newest-first by the API
            fam = _family(m["id"])
            if fam not in seen:
                seen.add(fam)
                filtered.append({"id": m["id"], "label": m["display_name"], "detail": ""})

        _models_cache = filtered
        log.info("fetch_anthropic_models: fetched %d models (%d after dedup) from API",
                 len(all_models), len(filtered))
        return _models_cache
    except Exception as exc:
        log.warning("fetch_anthropic_models: API call failed (%s), using fallback list", exc)
        return fallback

async def fetch_anthropic_oauth_usage() -> dict | None:
    """Fetch the live Claude subscription usage windows (5-hour + weekly).

    OAuth-subscription only — this endpoint authenticates with the keychain
    Bearer token (Pro/Max users) and rejects `x-api-key`. Returns the raw
    payload (`five_hour` / `seven_day` / `limits` / …) or None when no OAuth
    token is available or the call fails (expired token, offline, 4xx/5xx).

    Cached for `_OAUTH_USAGE_TTL` seconds — `/status` and `/usage` can be hit
    repeatedly and the windows only tick once every few minutes.
    """
    import asyncio, time
    global _oauth_usage_cache
    now = time.monotonic()
    if _oauth_usage_cache is not None and (now - _oauth_usage_cache[0]) < _OAUTH_USAGE_TTL:
        return _oauth_usage_cache[1]

    loop = asyncio.get_event_loop()
    token = await loop.run_in_executor(None, _keychain_oauth_token)
    if not token:
        _oauth_usage_cache = (now, None)
        return None

    headers = {
        "Authorization": f"Bearer {token}",
        "anthropic-beta": _ANTHROPIC_OAUTH_BETA,
        "anthropic-version": _ANTHROPIC_API_VERSION,
    }
    try:
        req = urllib.request.Request(_ANTHROPIC_OAUTH_USAGE_URL, headers=headers)
        data = await loop.run_in_executor(
            None, lambda: json.loads(urllib.request.urlopen(req, timeout=8).read())
        )
        _oauth_usage_cache = (now, data)
        return data
    except Exception as exc:
        log.warning("fetch_anthropic_oauth_usage: call failed (%s)", exc)
        _oauth_usage_cache = (now, None)
        return None


def _fmt_reset(resets_at: str) -> str:
    """Render an ISO8601 reset timestamp as 'in 2h 5m' (<24h) or 'Jun 27 06:00'."""
    from datetime import datetime, timezone
    try:
        dt = datetime.fromisoformat(resets_at.replace("Z", "+00:00"))
    except Exception:
        return ""
    delta = (dt - datetime.now(timezone.utc)).total_seconds()
    if delta <= 0:
        return "now"
    if delta < 86400:
        h, m = int(delta // 3600), int((delta % 3600) // 60)
        return f"in {h}h {m}m" if h else f"in {m}m"
    return dt.astimezone().strftime("%b %-d %H:%M")


def anthropic_usage_rows(usage: dict | None) -> list[dict]:
    """Map an /api/oauth/usage payload to {label, value} status rows.

    Reads the structured `limits` array (session / weekly_all / weekly_scoped),
    so model-scoped weekly caps surface under their model name. Empty list when
    there's no usable data.
    """
    if not usage:
        return []
    rows: list[dict] = []
    for lim in usage.get("limits") or []:
        pct = lim.get("percent")
        if pct is None:
            continue
        kind = lim.get("kind", "")
        if kind == "session":
            label = "5-hour limit"
        elif kind == "weekly_all":
            label = "Weekly (all models)"
        elif kind == "weekly_scoped":
            model = ((lim.get("scope") or {}).get("model") or {}).get("display_name") or "scoped"
            label = f"Weekly ({model})"
        else:
            label = kind.replace("_", " ").title()
        reset = _fmt_reset(lim.get("resets_at", ""))
        sev = lim.get("severity", "")
        flag = " ⚠️" if sev in ("critical", "warning") else ""
        value = f"{int(round(pct))}% used"
        if reset:
            value += f" · resets {reset}"
        rows.append({"label": label, "value": value + flag})
    return rows


_GEMINI_MODELS_URL = "https://generativelanguage.googleapis.com/v1beta/models?pageSize=200"
_gemini_models_cache: list[dict] | None = None


async def fetch_gemini_models(fallback: list[dict]) -> list[dict]:
    """Fetch the live Gemini model list from the public Generative Language API.

    Auth: a GEMINI_API_KEY / GOOGLE_API_KEY env var. The gemini-cli OAuth token
    (Code Assist scopes) is NOT sufficient — models.list returns 403 for it — so
    OAuth-only users fall back to the curated `fallback` list. Cached per process.
    """
    global _gemini_models_cache
    if _gemini_models_cache is not None:
        return _gemini_models_cache

    api_key = os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY", "")
    if not api_key:
        log.debug("fetch_gemini_models: no API key, using fallback list")
        return fallback

    try:
        import asyncio, re as _re
        req = urllib.request.Request(_GEMINI_MODELS_URL, headers={"x-goog-api-key": api_key})
        data = await asyncio.get_event_loop().run_in_executor(
            None, lambda: json.loads(urllib.request.urlopen(req, timeout=10).read())
        )
        # Keep chat-capable gemini models only; drop embedding/tts/imagen/aqa.
        _skip = ("embedding", "aqa", "tts", "image", "imagen", "vision")
        _ver_re = _re.compile(r"-(latest|preview|exp.*|\d{3,})$")

        def _family(mid: str) -> str:
            return _ver_re.sub("", mid)

        seen: set[str] = set()
        models: list[dict] = []
        for m in data.get("models", []):
            mid = (m.get("name", "") or "").removeprefix("models/")
            methods = m.get("supportedGenerationMethods") or m.get("supportedActions") or []
            if not mid.startswith("gemini") or "generateContent" not in methods:
                continue
            if any(s in mid for s in _skip):
                continue
            fam = _family(mid)
            if fam in seen:
                continue
            seen.add(fam)
            models.append({"id": mid, "label": m.get("displayName", mid), "detail": ""})

        if not models:
            return fallback
        _gemini_models_cache = models
        log.info("fetch_gemini_models: fetched %d gemini chat models from API", len(models))
        return _gemini_models_cache
    except Exception as exc:
        log.warning("fetch_gemini_models: API call failed (%s), using fallback list", exc)
        return fallback


_or_models_cache: list[dict] | None = None


async def fetch_openrouter_models(api_key: str) -> list[dict]:
    """Fetch the live OpenRouter model list.

    Returns [{id, label, detail}] sorted by the API's default order.
    Cached per process — the OR catalogue changes infrequently and the 1-hour
    _MODEL_CACHE_TTL in headless already caps re-fetch frequency.
    """
    global _or_models_cache
    if _or_models_cache is not None:
        return _or_models_cache

    try:
        import asyncio
        req = urllib.request.Request(
            "https://openrouter.ai/api/v1/models",
            headers={"Authorization": f"Bearer {api_key}"},
        )
        data = await asyncio.get_event_loop().run_in_executor(
            None, lambda: json.loads(urllib.request.urlopen(req, timeout=10).read())
        )
        models = []
        for m in data.get("data", []):
            mid = m.get("id", "")
            if not mid:
                continue
            label = m.get("name") or mid.split("/")[-1]
            models.append({"id": mid, "label": label, "detail": "via OpenRouter"})
        if models:
            _or_models_cache = models
            log.info("fetch_openrouter_models: fetched %d models", len(models))
        return models
    except Exception as exc:
        log.warning("fetch_openrouter_models: API call failed (%s)", exc)
        return []


def openrouter_fallback_models() -> list[dict]:
    """Hardcoded last-resort model list ({id, label, detail}) for adapters that
    route through OpenRouter (pi, hermes, …).

    Used only when neither the agent's own model probe nor a live OpenRouter
    fetch (shared provider key) yields anything — so the picker is never empty.
    The ids are `openrouter/<provider>/<model>`; selecting one still requires a
    working OpenRouter key at send time.
    """
    out: list[dict] = []
    for model_str, detail in (
        ("openrouter/anthropic/claude-sonnet-4.5", "via OpenRouter"),
        ("openrouter/anthropic/claude-opus-4.1",   "via OpenRouter"),
        ("openrouter/openai/gpt-5",                "via OpenRouter"),
        ("openrouter/openai/gpt-4o",               "via OpenRouter"),
        ("openrouter/google/gemini-2.5-pro",       "via OpenRouter"),
        ("openrouter/x-ai/grok-4",                 "via OpenRouter"),
        ("openrouter/moonshotai/kimi-k2.6",        "via OpenRouter"),
        ("openrouter/openai/gpt-4o-mini",          "via OpenRouter (cheap)"),
    ):
        out.append({"id": model_str, "label": model_str.rsplit("/", 1)[-1], "detail": detail})
    return out


# ── Shared inference providers (one global key → many models) ─────────────────
# A "shared provider" is a meta-endpoint whose single key unlocks many models
# across agents — OpenRouter today; Groq / Together / a self-hosted LiteLLM proxy
# are just more registry entries later (same discovery + policy). Discovery sweeps
# every place a key plausibly already lives, validates by key *shape* (so we never
# hand an agent a wrong-provider key), and records provenance for the UI.

from enum import Enum
from dataclasses import dataclass as _dataclass

# OpenRouter keys look like `sk-or-v1-<hex>` (older keys: `sk-or-<hex>`).
_OPENROUTER_KEY_RE = re.compile(r"sk-or-(?:v1-)?[A-Za-z0-9_-]{16,}")


@_dataclass(frozen=True)
class ProviderKey:
    """A discovered shared-provider key + where it came from."""
    key: str
    provider: str                     # registry id ("openrouter", "custom")
    source: str                       # "env" | "hermes-pool" | "opencode-auth" | …
    base_url: str = ""                # OpenAI-compatible /v1 root; "" = provider default


def _read_json(path: Path) -> dict | None:
    try:
        return json.loads(path.read_text(errors="replace"))
    except Exception:
        return None


def _openrouter_candidates() -> "list[tuple[str, str | None]]":
    """(source_label, raw_candidate) probes, in priority order. Lazy — each is
    cheap and guarded; the first shape-valid hit wins in discover_openrouter_key."""
    home = Path.home()
    out: list[tuple[str, str | None]] = []

    # 1. env (the canonical override)
    for var in ("OPENROUTER_API_KEY", "OPENROUTER_KEY"):
        out.append(("env", os.environ.get(var)))

    # 2. Beacon's own bridge env file
    env_file = home / ".config" / "terminull" / ".env"
    if env_file.is_file():
        for line in env_file.read_text(errors="replace").splitlines():
            if line.strip().startswith("OPENROUTER_API_KEY="):
                out.append(("beacon-config", line.split("=", 1)[1].strip().strip('"').strip("'")))

    # 3. Hermes credential pool (lowest `priority` wins)
    data = _read_json(home / ".hermes" / "auth.json") or {}
    pool = (data.get("credential_pool", {}) or {}).get("openrouter", []) or []
    for cred in sorted(pool, key=lambda c: c.get("priority", 0)):
        out.append(("hermes-pool", cred.get("access_token")))

    # 4. opencode's own auth (provider map: {"openrouter": {"type","key"}})
    for p in (home / ".local/share/opencode/auth.json",
              home / ".config/opencode/auth.json"):
        oc = (_read_json(p) or {}).get("openrouter")
        if isinstance(oc, dict):
            out.append(("opencode-auth", oc.get("key") or oc.get("access_token")))

    # 5. well-known single-key files
    for p in (home / ".config/openrouter/key", home / ".openrouter"):
        if p.is_file():
            out.append(("well-known", p.read_text(errors="replace").strip()))

    return out


_NOT_PROBED = object()                       # cache sentinel: discovery not yet run
_openrouter_key_cache: object = _NOT_PROBED


def discover_openrouter_key() -> ProviderKey | None:
    """Find an OpenRouter key the user already has, anywhere — env, Beacon config,
    the Hermes pool, opencode's auth, or a well-known file. Shape-validated and
    cached (model-list hot path). Read-only: we never copy a key between tools."""
    global _openrouter_key_cache
    if _openrouter_key_cache is not _NOT_PROBED:
        return _openrouter_key_cache  # type: ignore[return-value]
    found: ProviderKey | None = None
    for source, cand in _openrouter_candidates():
        if cand and _OPENROUTER_KEY_RE.fullmatch(cand.strip()):
            found = ProviderKey(key=cand.strip(), provider="openrouter", source=source)
            break
    _openrouter_key_cache = found
    return found


def reset_shared_provider_cache() -> None:
    """Drop the discovery cache so the next lookup re-probes — call when a key may
    have changed (a detection sweep, or after the user enters one)."""
    global _openrouter_key_cache
    _openrouter_key_cache = _NOT_PROBED


def openrouter_key() -> str | None:
    """Back-compat shim — the bare key string (see discover_openrouter_key)."""
    pk = discover_openrouter_key()
    return pk.key if pk else None


def _dev_openrouter() -> bool:
    """Dev/test forced-consent for OPT_IN agents (e.g. opencode): use the shared
    OpenRouter key even without explicitly selecting an openrouter/* model."""
    return os.environ.get("TERMINULL_DEV_OPENROUTER", "").lower() in ("1", "true", "yes")


def _discover_custom_provider() -> ProviderKey | None:
    """The user-configured OpenAI-compatible shared endpoint — any aggregator or
    proxy speaking the OpenAI chat-completions API (LiteLLM, Groq, Together,
    vLLM, …). Read from `[providers] custom_base_url/custom_api_key` in
    config.toml (set via the Mac menu / config file). Unlike the OpenRouter
    probe chain, this is explicitly configured *for the bridge to lend*, so it
    is not gated by share_openrouter_key. No shape validation — key formats
    vary per aggregator; base_url presence is the consent signal."""
    try:
        from ..config import custom_provider
        base_url, api_key, _label = custom_provider()
    except Exception:
        return None
    if not (base_url and api_key):
        return None
    return ProviderKey(key=api_key.strip(), provider="custom", source="beacon-config",
                       base_url=base_url.rstrip("/"))


def custom_provider_label() -> str:
    """Display label for the custom endpoint ("via <label>" in the model picker)."""
    try:
        from ..config import custom_provider
        label = custom_provider()[2]
    except Exception:
        label = ""
    return label or "Custom endpoint"


_openai_compat_models_cache: "dict[str, list[dict]]" = {}


async def fetch_openai_compat_models(pk: ProviderKey) -> list[dict]:
    """Live model list from any OpenAI-compatible endpoint (`GET {base}/models`).

    Returns [{id, label, detail}] like fetch_openrouter_models; cached per
    base_url for the process (same rationale — headless's _MODEL_CACHE_TTL
    already caps re-fetch frequency)."""
    cached = _openai_compat_models_cache.get(pk.base_url)
    if cached is not None:
        return cached
    try:
        import asyncio
        req = urllib.request.Request(
            f"{pk.base_url}/models",
            headers={"Authorization": f"Bearer {pk.key}"},
        )
        data = await asyncio.get_event_loop().run_in_executor(
            None, lambda: json.loads(urllib.request.urlopen(req, timeout=10).read())
        )
        detail = f"via {custom_provider_label()}"
        models = [{"id": m["id"], "label": str(m["id"]).split("/")[-1], "detail": detail}
                  for m in data.get("data", []) if m.get("id")]
        if models:
            _openai_compat_models_cache[pk.base_url] = models
            log.info("fetch_openai_compat_models: fetched %d models from %s",
                     len(models), pk.base_url)
        return models
    except Exception as exc:
        log.warning("fetch_openai_compat_models: %s failed (%s)", pk.base_url, exc)
        return []


@_dataclass(frozen=True)
class ProviderSpec:
    """One shared-provider registry entry — everything generic code needs to
    discover, inject, and catalog a provider without provider-specific branches.

    `discover` is late-bound (a lambda calling the module-level function) so
    tests can keep monkeypatching e.g. base.discover_openrouter_key."""
    id: str                                          # model-id namespace ("openrouter/…")
    env_var: str                                     # key env var injected into agent spawns
    discover: "Callable[[], ProviderKey | None]"
    fetch_models: "Callable[[ProviderKey], Awaitable[list[dict]]]"


PROVIDER_REGISTRY: "dict[str, ProviderSpec]" = {
    "openrouter": ProviderSpec(
        id="openrouter", env_var="OPENROUTER_API_KEY",
        discover=lambda: discover_openrouter_key(),
        fetch_models=lambda pk: fetch_openrouter_models(pk.key),
    ),
    "custom": ProviderSpec(
        id="custom", env_var="OPENAI_API_KEY",
        discover=lambda: _discover_custom_provider(),
        fetch_models=lambda pk: fetch_openai_compat_models(pk),
    ),
}


class ORPolicy(str, Enum):
    """How an adapter relates to the shared OpenRouter key.

    PRIMARY  — OpenRouter is the agent's default provider (pi): use the shared key
               when no model, or an openrouter/* model, is selected.
    BYOK     — env-only passthrough so the agent's own BYOK `custom:` models route
               through OpenRouter (droid). Never forces OR as the provider.
    OPT_IN   — respect the agent's own login by default; use the shared key only on
               explicit consent = an openrouter/* model selected (or the dev flag).
    NONE     — never inject the shared key (claude/codex/cursor/gemini/hermes/kilo).
    """
    PRIMARY = "primary"
    BYOK = "byok"
    OPT_IN = "opt_in"
    NONE = "none"


@_dataclass
class ProviderInjection:
    """The argv + env an adapter should add to use the shared provider, plus whether
    it actually applies (so HTTP-payload adapters like opencode can route too)."""
    active: bool = False
    provider_id: str | None = None
    env: dict = field(default_factory=dict)
    argv: list = field(default_factory=list)
    source: str | None = None
    base_url: str = ""                # custom endpoints: the OpenAI-compatible /v1 root


if TYPE_CHECKING:
    from ..attachments import Attachment


@dataclass
class ExternalSession:
    """A Claude/agent session started outside Beacon (e.g. directly from the CLI)."""
    session_id: str
    agent_id: str                     # "claude-code", "hermes", "kilo", ...
    cwd: str
    last_prompt: str = ""
    started_at: str = ""              # ISO-8601 if available
    is_active: bool = False           # True if the agent process is currently running
    # Liveness, finer-grained than is_active (drives the sidebar badge):
    #   "turn" — process alive AND transcript written in the last ~2s (generating)
    #   "open" — process alive, transcript quiet (idle at the prompt)
    #   "idle" — no live process (resumable)
    #   ""     — unknown; the collector falls back to is_active → open/idle
    # Scanners that can't measure write-velocity may leave this "" and just set is_active.
    activity: str = ""


@dataclass
class DetectedAgent:
    """An agent found on this machine — the result of BaseAdapter.detect()."""
    id: str                          # "claude-code", "cursor", "pi", ...
    name: str
    binary: str | None = None
    version: str = "unknown"
    mcp_tools: list[dict] = field(default_factory=list)
    extra: dict = field(default_factory=dict)
    available: bool = True           # False = installed but not responding
    available_models: list[dict] = field(default_factory=list)  # [{id, label, detail}]
    working_dirs: list[str] = field(default_factory=list)       # known project dirs
    beta: bool = True                # show a "Beta" badge in the app (set from the adapter class)
    # Credential readiness — installed (binary present) is not the same as usable.
    # ready=False ⇒ detected but no credential to run a turn (the app shows a
    # "Setup needed" state instead of letting a turn silently fail). setup_hint is
    # a short actionable note (shown when not ready, or to point at an enriching key).
    ready: bool = True
    setup_hint: str = ""
    model_scope: str = "list"   # "fixed"|"list"|"opt_in_or"|"any_or" — sent to iOS for picker behaviour


def probe_binary(binary: str | None) -> bool:
    """Return True if the binary runs and exits cleanly (`--version` probe)."""
    if not binary:
        return False
    try:
        r = subprocess.run([binary, "--version"], capture_output=True, timeout=5)
        return r.returncode == 0
    except Exception:
        return False


def own_subtree_pids() -> set[int]:
    """PIDs of the bridge process and all its descendants (via ps ppid chains).

    The session-handoff ownership scanners must exclude Beacon's *own* spawned agent
    subprocesses (claude/codex/cursor/… launched by an adapter) — otherwise a live
    Beacon session's own subprocess looks like an external native CLI "owning" the
    cwd, which wrongly parks sibling Beacon sessions in the same directory as detached
    observers (and makes "Continue in Terminal" a no-op). A genuine native terminal CLI
    is NOT a descendant of the bridge, so it's still detected. See docs/session-handoff.md."""
    import os as _os
    try:
        ps = subprocess.check_output(["ps", "-axo", "pid=,ppid="], text=True,
                                     stderr=subprocess.DEVNULL)
    except Exception:
        return set()
    children: dict[int, list[int]] = {}
    for line in ps.splitlines():
        parts = line.split()
        if len(parts) != 2:
            continue
        try:
            pid, ppid = int(parts[0]), int(parts[1])
        except ValueError:
            continue
        children.setdefault(ppid, []).append(pid)
    seen: set[int] = set()
    stack = [_os.getpid()]
    while stack:
        p = stack.pop()
        if p in seen:
            continue
        seen.add(p)
        stack.extend(children.get(p, ()))
    return seen


# Substrings that appear ONLY in a Beacon-spawned agent CLI invocation, never in a
# user's native terminal session. Used to exclude our own subprocesses from
# external-ownership detection even when they are NOT in our process subtree — e.g.
# orphaned subprocesses left by a *previous* bridge instance after a restart, which
# get reparented to launchd/init and would otherwise look like a native CLI "owning"
# the cwd, falsely parking restored sessions as detached observers. See docs/session-handoff.md.
_TERMINULL_SPAWN_MARKERS = ("beacon_mcp_", "Beacon session guidelines", "Terminull session guidelines")


def is_beacon_spawned_cmdline(cmdline: str) -> bool:
    """True if a process command line carries a Beacon-injected signature (our MCP
    config temp path or appended system prompt) — i.e. we spawned it, not the user."""
    return any(m in cmdline for m in _TERMINULL_SPAWN_MARKERS)


def active_cwds_for_command(needle: str) -> set[str]:
    """Working dirs of every running process whose command line contains `needle`.

    Matches the full command (`ps -axo pid,command`) not just `comm`, so it works
    for node/bash-wrapped CLIs (e.g. cursor-agent) where `comm` would be "node".
    Beacon's own spawned subprocesses are excluded — both by process subtree
    (`own_subtree_pids`) and by command-line signature (`is_beacon_spawned_cmdline`,
    which also catches orphans from a prior bridge instance after a restart).
    Used by the session-handoff cwd-ownership checks. See docs/session-handoff.md."""
    import os as _os
    out: set[str] = set()
    try:
        ps = subprocess.check_output(["ps", "-axo", "pid=,command="], text=True,
                                     stderr=subprocess.DEVNULL)
    except Exception:
        return out
    own = own_subtree_pids()  # never count our own agent subprocesses as "external"
    for line in ps.splitlines():
        line = line.strip()
        if not line or needle not in line:
            continue
        if is_beacon_spawned_cmdline(line):
            continue
        pid = line.split(None, 1)[0]
        if not pid.isdigit():
            continue
        if int(pid) in own:
            continue
        try:
            lsof = subprocess.check_output(["lsof", "-p", pid, "-a", "-d", "cwd", "-Fn"],
                                           text=True, stderr=subprocess.DEVNULL)
        except Exception:
            continue
        for ln in lsof.splitlines():
            if ln.startswith("n/") and _os.path.isdir(ln[1:]):
                out.add(ln[1:])
    return out


def external_pids_in_cwd(needle: str, cwd: str, *, resume_sid: str = "") -> list[int]:
    """PIDs of *non-Beacon* processes whose command contains `needle` and whose cwd is
    `cwd`. Excludes the bridge's own subtree (`own_subtree_pids`), so it returns only a
    genuine native terminal CLI. Used by "smart Take over" to terminate the native agent
    so the app can reclaim the session. See docs/session-handoff.md.

    `resume_sid`: when set, only return processes whose cmdline contains
    `--resume <resume_sid>` — so "Take Over" kills only the one session that
    conflicts, not every claude process in the directory."""
    import os as _os
    if not cwd:
        return []
    target = _os.path.realpath(_os.path.expanduser(cwd))
    own = own_subtree_pids()
    pids: list[int] = []
    try:
        ps = subprocess.check_output(["ps", "-axo", "pid=,command="], text=True,
                                     stderr=subprocess.DEVNULL)
    except Exception:
        return pids
    for line in ps.splitlines():
        line = line.strip()
        if not line or needle not in line:
            continue
        if is_beacon_spawned_cmdline(line):  # never kill our own (or orphaned) subprocess
            continue
        spid = line.split(None, 1)[0]
        if not spid.isdigit() or int(spid) in own:
            continue
        # When a specific session is targeted, only match the process resuming it.
        if resume_sid and f"--resume {resume_sid}" not in line:
            continue
        try:
            lsof = subprocess.check_output(["lsof", "-p", spid, "-a", "-d", "cwd", "-Fn"],
                                           text=True, stderr=subprocess.DEVNULL)
        except Exception:
            continue
        for ln in lsof.splitlines():
            if ln.startswith("n/") and _os.path.realpath(ln[1:]) == target:
                pids.append(int(spid))
                break
    return pids

# Bridge-internal secrets that must never be inherited by agent subprocesses.
_BRIDGE_SECRET_VARS: frozenset[str] = frozenset({
    "TERMINULL_AUTH_TOKEN",
    "TERMINULL_RELAY_URL",
    "TERMINULL_API_URL",
})

# Injected into every session's cwd when no CLAUDE.md exists yet.
_SESSION_CLAUDE_MD = """\
# Terminull session rules
- Stay within this working directory; do not access parent directories.
- Do not read bridge internals: ~/.terminull/, ~/beacon/bridge/, or any bridge config.
- Do not attempt to modify bridge settings or restart the bridge daemon.
- Use the terminull-admin CLI for any bridge-level operations.
"""


def build_append_system_prompt(beacon_id: str | None = None) -> str:
    """Build a dynamic system prompt appendix injected at session start.

    Includes:
    - Instruction to ask the user via request_user_input before proceeding when intent is unclear.
    - Beacon file link format instruction (when beacon_id is provided).
    - Restricted terminal paths from config (if any).
    """
    from ..config import get_terminal_restricted_paths
    lines: list[str] = [
        "## Terminull session guidelines",
        "",
        "**Ask before acting when intent is unclear.** If the user's request is ambiguous, you are missing key information, or an action is risky, "
        "use the `request_user_input` MCP tool to ask a focused question before proceeding. "
        "Do not guess or make assumptions — one well-placed question prevents wasted work.",
        "",
        "**Proactively offer choices.** When there are multiple reasonable approaches to a task, "
        "use `request_user_input` with `input_type=choice` to let the user pick rather than silently choosing for them. "
        "Also use it to confirm scope before large or irreversible changes, and to gather preferences "
        "(e.g. naming conventions, library choices, tone of output). "
        "A short question upfront saves a full redo later.",
    ]
    if beacon_id:
        lines += [
            "",
            f"**Sharing local files with the user.** To let the user preview a file from this machine, "
            f"include a Markdown link using the format `https://{beacon_id}.terminull/absolute/path/to/file`. "
            f"Example: `[arch.md](https://{beacon_id}.terminull/Users/example/project/docs/arch.md)`. "
            f"Tapping the link in the Terminull app opens a local preview — no upload or external URL needed.",
        ]
    restricted = get_terminal_restricted_paths()
    if restricted:
        lines += [
            "",
            "**Do not open terminals in these restricted directories** (or any of their subdirectories):",
        ]
        for p in restricted:
            lines.append(f"- `{p}`")
        lines.append(
            "If the user asks to open a terminal in a restricted path, "
            "explain that it is off-limits and suggest an alternative."
        )
    lines += ["", build_beacon_tools_preprompt()]
    lines += _custom_prompt_lines()
    return "\n".join(lines)


def _custom_prompt_lines() -> list[str]:
    """The operator's append-only custom prompt block, or [] when the per-machine
    `[prompts] use_custom_prompts` toggle is off / the text is empty. Appended LAST —
    after the built-in guidelines + tools contract — so a custom addition can steer the
    agent but can't strip the rich-app tool wiring above it. Shared by every injection
    path (build_append_system_prompt for claude + sentinel adapters, hermes instructions)."""
    try:
        from ..config import effective_custom_prompt
        text = effective_custom_prompt()
    except Exception:
        text = ""
    if not text:
        return []
    return [
        "",
        "## Custom instructions (from this machine's operator)",
        text,
    ]


def build_beacon_tools_preprompt() -> str:
    """The shared, capability-gated nudge describing the Beacon (sentinel) MCP tools.

    The tools are auto-discovered by any agent we registered the MCP into, so this is
    the complementary "when/why to call them" guidance. Inject it ONLY for adapters
    with `cap_sentinel = True` (the tools exist + route there) — never elsewhere, or
    the agent will hallucinate calls to tools it doesn't have.
    """
    return "\n".join([
        "## Terminull tools (mcp server `terminull`)",
        "You are talking to a user through the **Terminull iOS app** — a rich chat client, "
        "not a plain terminal. It renders your replies as formatted Markdown with syntax-highlighted "
        "code, collapsible shell/command output, and inline previews of images and files. So share "
        "results visually: link local files for an in-app preview (see the file-sharing note above) "
        "and use the native tools below rather than pasting raw paths, bare URLs, or dumping long output as text.",
        "You have native Terminull tools — prefer them over plain text when they fit:",
        "- `request_user_input` — ask the user a focused question (supports `input_type=choice`).",
        "- `request_permission` — get approval BEFORE a destructive/irreversible action.",
        "- `suggest_next_steps` — offer tappable follow-up chips after a turn.",
        "- `open_url` — show a rich, tappable link-preview card (OpenGraph banner/title/site) in chat; the user taps it to open in an in-app browser. This is the DEFAULT way to surface any link to the user — prefer it over pasting a bare URL in your reply. Whenever you reference a link the user might open (docs, a PR/issue, a repo, a deploy/preview URL, an article), send it through `open_url` instead of writing the raw URL in prose — not only when asked. Gentle and non-intrusive. NOT for fetching pages yourself — use your own fetch/browse tool for that.",
        "- `open_terminal` — open an interactive terminal for the USER to type in (user-facing only; NOT for running your own commands — use your built-in shell for that). Use only when the user explicitly asks to open a terminal/shell.",
        "- `request_file_download` — share a local file INLINE as a tappable media preview bubble (image thumbnail, video plays inline, else a preview + download chip). The way to SHOW the user a file you made — a screenshot, chart, video, PDF — not by pasting a path or raw link.",
        "- `show_photo` — display an image from a local path inline as a tappable photo bubble (images only; for video or any other file use `request_file_download`).",
        "- `create_plugin` (alias `suggest_command`) — save a reusable one-tap PLUGIN in the composer's plugins bar. A plugin either runs a shell command on this machine (a status check / build / deploy — output shown inline or as a live card) OR sends a canned prompt to the agent; these tools create the shell-command kind. Offer one after a flow the user will repeat — including a manual trigger for something otherwise on a schedule. Emits a chip the user taps to install; never creates it silently.",
        "- `suggest_schedule` — set up a recurring run when you spot a periodic pattern (a daily digest, an hourly check). A schedule is either an AGENT run — an LLM turn on your prompt each fire (the default) — or a deterministic SCRIPT run (a shell command; exit 0 notifies, 75 stays silent). Emits a tap-to-create chip.",
        "- `notify_user` — reach the user out-of-band (a phone push + a persistent inbox card) when a long task finishes, a check/threshold trips, or a decision blocks you — i.e. when they're away. NOT for chatter in a turn they're already reading.",
        "Call them by their bare names; don't print fake tool output. If a tool isn't "
        "available in this session, just proceed without it.",
    ])


def agent_env(extra: dict[str, str] | None = None, *, session_id: str | None = None) -> dict[str, str]:
    """Return os.environ minus bridge secrets, plus any caller-supplied additions.

    Pass extra={} for any env vars the agent legitimately needs beyond the default set.

    Pass session_id to stamp TERMINULL_SESSION_ID into the subprocess env. The globally-
    registered sentinel MCP script (~/.terminull/sentinel_mcp.py) reads it to route the
    agent's beacon tool calls back to the right Beacon session. Each adapter instance
    owns exactly one session/server, so a per-spawn value is correct (opencode/pi);
    fresh-per-turn adapters (codex/gemini/cursor/droid) inherit it too, ready for when
    their MCP is registered. Claude routes via its own per-session --mcp-config instead.
    """
    env = {k: v for k, v in os.environ.items() if k not in _BRIDGE_SECRET_VARS}
    if extra:
        env.update(extra)
    if session_id:
        env["TERMINULL_SESSION_ID"] = session_id
    return env


def write_session_claude_md(cwd: str | None) -> None:
    """Write a CLAUDE.md with session isolation rules to cwd, if none exists yet.

    Opt-in: only runs when TERMINULL_INJECT_CLAUDE_MD=1 is set in the environment.
    """
    if not cwd or os.environ.get("TERMINULL_INJECT_CLAUDE_MD") != "1":
        return
    path = Path(cwd) / "CLAUDE.md"
    if not path.exists():
        path.write_text(_SESSION_CLAUDE_MD)

_FENCE_RE = re.compile(r"```(\w*)\n(.*?)```", re.DOTALL)

# Commands the bridge handles universally across all adapters.
# Declared in every AgentInfoBlock so iOS shows them in the picker.
# /stream-test + /stream-test2 are dev-only streaming stress commands — never exposed on a
# headless/public bridge. Gated on TERMINULL_DEV (legacy BEACON_DEV honored too); the handlers
# in session.py stay, they just aren't advertised unless the dev flag is set.
_DEV_MODE = os.environ.get("TERMINULL_DEV", os.environ.get("BEACON_DEV", "")).strip().lower() in ("1", "true", "yes", "on")
BRIDGE_SLASH_COMMANDS = (
    ["/stop", "/clear", "/tools", "/status", "/diff", "/cost", "/usage", "/ask"]
    + (["/stream-test", "/stream-test2", "/stream-batch"] if _DEV_MODE else [])
    + ["/all-sessions"]
)

_TOOL_LABELS: dict[str, str] = {
    "Bash": "bash", "Read": "read", "Write": "write",
    "Edit": "edit", "MultiEdit": "edit", "Grep": "grep", "Glob": "glob",
    "WebSearch": "search", "WebFetch": "fetch",
    # Anthropic's native server-side tool names (content_block type "server_tool_use"),
    # distinct from the client-tool "WebSearch"/"WebFetch" names above.
    "web_search": "search", "web_fetch": "fetch",
    "Agent": "agent", "Task": "task", "TodoWrite": "todo",
}
_PATH_KEYS = ("file_path", "path", "id")
_TEXT_KEYS  = ("command", "description", "pattern", "query", "url", "action", "prompt")

# Tools that mutate a file on disk. Used by Session to track which files a turn
# touched (for the turn-end FileChangesBlock). Bash-driven writes are caught
# separately via `git status` at turn end.
_MUTATING_TOOLS = frozenset({"Write", "Edit", "MultiEdit", "NotebookEdit"})


def mutated_path(name: str, args: dict) -> str | None:
    """Absolute path a mutating tool wrote to, or None if not a file mutation."""
    if name not in _MUTATING_TOOLS:
        return None
    for key in ("file_path", "path", "notebook_path"):
        if val := args.get(key):
            return str(val)
    return None

def tool_label_subject(name: str, args: dict) -> tuple[str, str]:
    """Return (label, subject) display hints for a tool call."""
    label = _TOOL_LABELS.get(name) or (
        name.split("__")[-1] if name.startswith("mcp__") else name
    )
    subject = ""
    for key in _PATH_KEYS:
        if val := args.get(key):
            subject = str(val).rsplit("/", 1)[-1]
            break
    if not subject:
        for key in _TEXT_KEYS:
            if val := args.get(key):
                subject = str(val)[:80].split("\n")[0]
                break
    return label, subject


def split_markdown(text: str) -> list[Block]:
    """Split assistant text into alternating TextBlock / CodeBlock pieces."""
    result: list[Block] = []
    last = 0
    for m in _FENCE_RE.finditer(text):
        pre = text[last:m.start()]
        if pre.strip():
            result.append(TextBlock(text=pre, id=str(uuid.uuid4())))
        result.append(CodeBlock(content=m.group(2), language=m.group(1), id=str(uuid.uuid4())))
        last = m.end()
    tail = text[last:]
    if tail.strip():
        result.append(TextBlock(text=tail, id=str(uuid.uuid4())))
    return result or [TextBlock(text=text, id=str(uuid.uuid4()))]


class BaseAdapter(ABC):
    """One adapter instance = one interactive session with one agent.

    Subclassing contract (enforced at import via ``__init_subclass__`` — see
    below — so a new adapter cannot silently inherit a no-op):

    1. Implement the four abstract coroutines: ``start`` / ``send`` /
       ``stream`` / ``stop``.
    2. Declare capabilities as ``cap_*`` class attributes (the single source of
       truth). ``make_agent_info()`` reads them into the ``AgentInfoBlock`` —
       adapters should build their info block through that helper rather than
       hand-rolling caps + slash commands.
    3. Declare, explicitly, whether the adapter provides the two optional
       surfaces below by setting these class flags to ``True`` or ``False``
       (there is no default — "this adapter does nothing here" must be a
       visible, deliberate choice, not an inherited blank):

           provides_models  → must override ``list_models``
           provides_usage   → must override ``get_usage``

       Setting a flag ``True`` without overriding the method (or leaving a flag
       unset on a concrete adapter) raises ``TypeError`` at import time.
    """

    # ── Capability contract — declare as class attributes, emitted via
    #    make_agent_info(). Names mirror AgentInfoBlock.cap_* fields. ──────────
    cap_images: bool = False             # image input accepted
    cap_documents: bool = False          # non-image file input (PDFs, …) — attachment pipeline
    cap_thinking: bool = False           # emits thinking/reasoning blocks
    cap_permissions: bool = False        # emits permission_request blocks
    cap_subagents: bool = False          # emits Agent/subagent tool calls
    cap_interrupt: bool = False          # supports mid-turn stop
    cap_input_during_turn: bool = False  # accepts a new message while generating
    cap_terminal: bool = True            # may open a terminal on the iOS device
    cap_sentinel: bool = False           # beacon sentinel MCP tools registered + routable (P2/P3)
    # Persistent-process adapters (one subprocess across turns) that may self-exit
    # cleanly (rc=0) between turns: opt in so the Session eager-respawns the process in
    # the background (while a client is connected) instead of paying a cold restart on
    # the NEXT message's critical path. Per-turn adapters (spawn fresh each turn) MUST
    # leave this False — they exit rc=0 by design and must not be kept warm.
    eager_respawn_on_exit: bool = False

    # ── Optional-surface declarations (explicit opt-out, no default) ─────────
    provides_models: bool | None = None  # implements list_models()
    provides_usage: bool | None = None   # implements get_usage()

    # ── Detection identity (used by detect() + the agent registry) ───────────
    agent_id: str | None = None          # stable registry id ("cursor", "pi", …)
    display_name: str | None = None      # shown in the iOS agent picker
    cli_binary: str | None = None        # PATH binary the default detect() probes
    beta: bool = True                    # "Beta" badge in the app; flip to False once
    #                                      live-verified (e.g. passing the adapter matrix)

    # The owning Beacon session id, stamped in by ``start(session_id)`` and read
    # when spawning agent subprocesses (routes the sentinel MCP back to this
    # session — see ``agent_env``). Declared here so adapters read ``self.
    # _beacon_session_id`` directly instead of a ``getattr(self, …, "")`` shim.
    _beacon_session_id: str = ""
    # The owning bridge's beacon id (server-assigned, per-bridge constant). Stamped in by
    # headless._make_adapter after construction so every adapter — not just claude — can
    # build the `https://{beacon_id}.terminull/…` file-preview link in its client-context
    # preamble. Empty on probes / ext adapters that were never stamped.
    _beacon_id: str = ""
    # One-shot latch for the sentinel client-context preamble (persistent adapters:
    # pi/opencode/kilo, whose send() runs every turn). Set True after the first send so
    # the preamble rides only the first user message of this process's lifetime.
    _sentinel_preamble_sent: bool = False
    _cwd: "str | None" = None            # working dir; the ``cwd`` property reads it
    _proc: "asyncio.subprocess.Process | None" = None  # live subprocess (CLI adapters)

    # ── Shared-provider policy ────────────────────────────────────────────────
    or_policy: ORPolicy = ORPolicy.NONE  # how this adapter uses shared provider keys
    # Which PROVIDER_REGISTRY entries this adapter can actually route through.
    # OpenRouter is universal; add "custom" only once the adapter maps a generic
    # OpenAI-compatible endpoint into its own provider config (see goose).
    shared_providers: "tuple[str, ...]" = ("openrouter",)

    @classmethod
    def shared_provider_keys(cls) -> "list[ProviderKey]":
        """Every configured shared-provider key this adapter may *lend to its
        subprocess* — policy + gate + discovery, in `shared_providers` order,
        independent of per-turn routing consent (that's shared_provider_injection).
        The share_openrouter_key gate covers the *discovered* OpenRouter key only;
        a custom endpoint is configured on the bridge expressly to be lent."""
        if (cls.or_policy or ORPolicy.NONE) is ORPolicy.NONE:
            return []
        gate_open = True
        try:
            from ..config import share_openrouter_key
            gate_open = share_openrouter_key()
        except Exception:
            pass
        out: "list[ProviderKey]" = []
        for pid in cls.shared_providers:
            spec = PROVIDER_REGISTRY.get(pid)
            if not spec or (pid == "openrouter" and not gate_open):
                continue
            pk = spec.discover()
            if pk:
                out.append(pk)
        return out

    @classmethod
    def shared_provider_key(cls) -> "ProviderKey | None":
        """First configured shared key (back-compat single-provider view) — see
        shared_provider_keys()."""
        keys = cls.shared_provider_keys()
        return keys[0] if keys else None

    @classmethod
    def _model_scope(cls) -> str:
        """Picker behaviour the iOS app should apply for this adapter.

        "fixed"      — no model picker (provides_models=False)
        "list"       — show exactly what the bridge sends; no OR opt-in
        "opt_in_or"  — own provider login is the default; OR models are opt-in
        "any_or"     — OR is the primary provider; any OR model is valid
        """
        if not cls.provides_models:
            return "fixed"
        policy = cls.or_policy or ORPolicy.NONE
        if policy is ORPolicy.PRIMARY:
            return "any_or"
        if policy is ORPolicy.OPT_IN:
            return "opt_in_or"
        return "list"

    @classmethod
    def shared_provider_injection(cls, model: str | None) -> ProviderInjection:
        """The (argv, env) an adapter should add to run on a shared provider
        key, per its `or_policy`, the bridge gate, and key discovery. One place,
        so every adapter behaves identically — no bespoke `_use_openrouter` logic.

        The provider is picked by the model-id namespace (`openrouter/…`,
        `custom/…`) among this adapter's `shared_providers`. Consent model:
        PRIMARY defaults to the first configured provider (model unset) or the
        model's own; OPT_IN requires an explicitly-selected namespaced model
        (the picker is the consent) or the dev flag; BYOK only passes the
        key(s) through as env.
        """
        policy = cls.or_policy or ORPolicy.NONE
        keys = cls.shared_provider_keys()   # gate + policy(≠NONE) + discovery
        if not keys:
            return ProviderInjection()
        # The provider the selected model is namespaced under, if any.
        chosen = next((k for k in keys if model and model.startswith(k.provider + "/")), None)

        if policy is ORPolicy.BYOK:
            # Key(s) as env only; the agent selects its own custom:<id> models.
            env = {PROVIDER_REGISTRY[k.provider].env_var: k.key for k in keys}
            return ProviderInjection(active=True, provider_id=keys[0].provider,
                                     env=env, source=keys[0].source)

        if policy is ORPolicy.PRIMARY:
            pk = chosen or (keys[0] if model is None else None)
        else:  # OPT_IN — explicit consent only
            pk = chosen or (keys[0] if _dev_openrouter() else None)
        if not pk:
            return ProviderInjection()
        # `--provider openrouter` is a first-class flag on OR-native CLIs (pi);
        # a custom endpoint has no universal argv — adapters map base_url/env
        # into their own provider config (see goose._build_env).
        argv = ["--provider", pk.provider] if pk.provider == "openrouter" else []
        return ProviderInjection(active=True, provider_id=pk.provider,
                                 env={PROVIDER_REGISTRY[pk.provider].env_var: pk.key},
                                 argv=argv, source=pk.source, base_url=pk.base_url)

    @classmethod
    async def shared_catalog_models(cls) -> "list[dict]":
        """Model catalog across every configured shared provider, ids namespaced
        by provider (`openrouter/<slug>`, `custom/<slug>`). Falls back to the
        hardcoded OpenRouter list so a shared-provider picker is never empty."""
        out: "list[dict]" = []
        for pk in cls.shared_provider_keys():
            try:
                ms = await PROVIDER_REGISTRY[pk.provider].fetch_models(pk)
            except Exception as exc:
                log.warning("shared_catalog_models: %s fetch failed (%s)", pk.provider, exc)
                ms = []
            out += [{"id": f"{pk.provider}/{m['id']}", "label": m["label"],
                     "detail": m["detail"]} for m in ms]
        return out or openrouter_fallback_models()

    @classmethod
    def readiness(cls) -> "tuple[bool, str]":
        """(can_run_a_turn, short_actionable_hint).

        Being installed (binary on PATH) is not the same as being usable — most
        agents need a login/API key. Override per-adapter to check the agent's own
        credential and return a one-line hint when it (or an enriching key) is
        missing. Default: ready, no hint."""
        return True, ""

    @classmethod
    def detect(cls) -> "DetectedAgent | None":
        """Return a DetectedAgent if this agent is available on this machine, else None.

        This is the single detection entry point for **both** built-in and
        external adapters — `detector.detect_all()` just calls `cls.detect()`
        on every registered adapter class.

        Default behaviour: requires `agent_id`; if `cli_binary` is set the binary
        must be on PATH (probed with `--version`). Override for HTTP-, config-, or
        otherwise non-PATH-based detection (see claude_code/hermes/kilo/vscode).
        """
        if not cls.agent_id:
            return None
        binary = None
        if cls.cli_binary:
            binary = shutil.which(cls.cli_binary)
            if not binary:
                return None
        return DetectedAgent(
            id=cls.agent_id,
            name=cls.display_name or cls.agent_id,
            binary=binary,
            available=probe_binary(binary) if binary else True,
        )

    def __init_subclass__(cls, **kwargs) -> None:
        super().__init_subclass__(**kwargs)
        # Only concrete (instantiable) adapters carry the contract; skip
        # intermediate abstract bases that still have unimplemented methods.
        if inspect.isabstract(cls):
            return

        def _overrides(name: str) -> bool:
            sub = getattr(getattr(cls, name), "__func__", getattr(cls, name))
            base = getattr(getattr(BaseAdapter, name), "__func__", getattr(BaseAdapter, name))
            return sub is not base

        problems: list[str] = []
        for flag, method in (("provides_models", "list_models"),
                             ("provides_usage", "get_usage")):
            declared = getattr(cls, flag)
            if declared is None:
                problems.append(f"must set `{flag} = True`/`False`")
            elif declared and not _overrides(method):
                problems.append(f"`{flag} = True` but does not override `{method}()`")
        if problems:
            raise TypeError(
                f"{cls.__name__} violates the BaseAdapter contract: "
                + "; ".join(problems)
            )

    @abstractmethod
    async def start(self, session_id: str) -> AgentInfoBlock:
        """Connect/spawn the agent. Return capability info (via make_agent_info)."""

    @abstractmethod
    async def send(self, text: str, attachments: list["Attachment"] | None = None) -> None:
        """Send a user message (with optional file attachments) into the running session."""

    @abstractmethod
    def stream(self) -> AsyncIterator[Block]:
        """Yield normalized blocks as the agent produces them."""

    @abstractmethod
    async def stop(self) -> None:
        """Cleanly shut down the session."""

    def clear_history(self) -> None:
        """Reset so the NEXT turn starts a brand-new agent session with empty context.

        Contract (the universal /clear): forget/regenerate any resume id this adapter
        holds so the next spawn does NOT --resume the prior session — the agent then
        mints a fresh id. A persistent adapter that holds a live process must also
        restart it (set `_pending_restart` and end the stream) so the cleared context
        takes effect immediately rather than on the next turn.

        The Beacon `session_id` is the stable app-facing identity and MUST NOT change —
        only the underlying agent/ext session id is regenerated, so no client-side
        session remap is ever needed. Default is a no-op for adapters with no
        resumable state; every adapter that keeps a resume id overrides this.
        """

    # ── Shared helpers ───────────────────────────────────────────────────────

    def sentinel_client_preamble(self, is_first_turn: bool) -> str:
        """Client-context + Terminull-tools preprompt to PREPEND to the first user
        message, for `cap_sentinel` adapters whose CLI has no claude-style
        `--append-system-prompt` channel (codex/cursor/droid/goose/copilot/opencode/
        kilo/pi). It tells the agent it's talking to the rich Terminull iOS app and
        nudges it to use the sentinel MCP tools (file/image previews, choices, links)
        instead of dumping raw text — the same content claude gets out-of-band via
        `build_append_system_prompt` and hermes via its run `instructions`.

        Returns "" when not applicable (not the first turn, or not a sentinel adapter)
        so callers can prepend unconditionally. Only the first turn carries it; the
        agent's own session/resume memory carries it forward. Wrapped in a tagged block
        so the agent reads it as ambient client context, not as the user's words.
        """
        if not is_first_turn or not type(self).cap_sentinel:
            return ""
        body = build_append_system_prompt(beacon_id=(self._beacon_id or None))
        return f"<terminull-client-context>\n{body}\n</terminull-client-context>\n\n"

    def make_agent_info(self, *, name: str, version: str = "",
                        tools: list[dict] | None = None,
                        slash_commands: list[str] | None = None,
                        working_dirs: list[str] | None = None,
                        cwd: str = "",
                        effort_levels: list[str] | None = None,
                        effort: str = "",
                        **cap_overrides: bool) -> AgentInfoBlock:
        """Build the AgentInfoBlock from the adapter's declared cap_* attributes.

        BRIDGE_SLASH_COMMANDS are prepended automatically — pass `slash_commands`
        for agent-native commands only. `cap_*` keyword overrides win over the
        class declaration, for caps that are only known at runtime (e.g. pi's
        per-model image support, claude_api's thinking toggle).
        """
        def cap(field: str, default: bool) -> bool:
            return bool(cap_overrides.get(field, default))

        return AgentInfoBlock(
            name=name,
            version=version,
            tools=tools or [],
            slash_commands=list(BRIDGE_SLASH_COMMANDS) + list(slash_commands or []),
            working_dirs=working_dirs or [],
            cwd=cwd or self.cwd,
            cap_images=cap("cap_images", self.cap_images),
            cap_thinking=cap("cap_thinking", self.cap_thinking),
            cap_permissions=cap("cap_permissions", self.cap_permissions),
            cap_subagents=cap("cap_subagents", self.cap_subagents),
            cap_interrupt=cap("cap_interrupt", self.cap_interrupt),
            cap_input_during_turn=cap("cap_input_during_turn", self.cap_input_during_turn),
            cap_terminal=cap("cap_terminal", self.cap_terminal),
            cap_sentinel=cap("cap_sentinel", self.cap_sentinel),
            # Derived: an agent supports "Continue in Terminal" iff it provides a
            # terminal resume command (claude/pi/hermes do; editor/server agents don't).
            cap_continue_in_terminal=cap(
                "cap_continue_in_terminal", type(self).terminal_resume_command("probe") is not None),
            effort_levels=list(effort_levels or []),
            effort=effort,
        )

    def todo_block(self, session_id: str, raw_todos: list[dict],
                   *, id_prefix: str = "todos") -> TodoBlock:
        """Build the live TodoBlock from an agent's plan list.

        Plan-capable agents re-send the whole list each update; we emit a
        generic TodoBlock with a stable per-session id so iOS upserts one card
        in place. The opencode-family `cancelled` status has no TodoBlock
        equivalent → mapped to `completed` (no-op for agents that never emit it).

        `id_prefix` lets an adapter render a *second* independent card (e.g.
        Hermes subagent workstreams use `id_prefix="subagents"`) without
        clobbering its plan card.
        """
        return TodoBlock(
            id=f"{id_prefix}:{session_id}",
            items=[
                {"content": t.get("content", ""),
                 "status": ("completed" if t.get("status") == "cancelled"
                            else t.get("status", "pending"))}
                for t in raw_todos
            ],
        )

    @property
    def cwd(self) -> str:
        """Working directory for this session. Used for path enrichment in text blocks."""
        return getattr(self, '_cwd', '') or ''

    async def get_usage(self) -> dict:
        """Return accumulated usage stats for /status. Override when provides_usage=True."""
        return {}

    async def get_live_limits(self) -> list[dict]:
        """Return {label, value} rows for live account/plan limits (e.g. subscription
        usage windows), fetched on demand for /status and /usage. Empty by default."""
        return []

    async def get_status_hint(self) -> "Block | None":
        """Return current stats as a periodic hint block, or None if not available."""
        return None

    @classmethod
    def scan_external_sessions(cls) -> list[ExternalSession]:
        """Scan for agent sessions started outside Beacon. Override per-adapter."""
        return []

    @classmethod
    def is_cwd_externally_owned(cls, cwd: str, sid: str = "") -> bool:
        """True if this session is currently being driven outside Beacon.

        Used by the session-handoff contention guard: Beacon must not start a second
        writer into a session something else is driving. Most agents key this on `cwd`
        (a native process in that directory); session-keyed agents (e.g. a shared
        daemon) use `sid` instead and ignore `cwd`. Default False (no detection);
        override per-adapter. See docs/session-handoff.md."""
        return False

    @classmethod
    def latest_sid_for_cwd(cls, cwd: str) -> str | None:
        """Newest agent session id whose state lives under `cwd` (the lineage head).

        Used when attaching after a terminal hand-back: `--resume` can mint a new
        sid, so the head must be re-resolved from disk. Default None (use the stored
        sid); override per-adapter. See docs/session-handoff.md."""
        return None

    @classmethod
    def make_transcript_mirror(cls, cwd: str, sid: str, *,
                               from_end: bool = False) -> "object | None":
        """Return a transcript mirror (an object with `read_new_blocks() -> list[dict]`)
        for read-only viewing of a terminal-owned session, or None if this agent has no
        readable transcript. `from_end=True` (the release path) tails from EOF without
        backfilling. Default None; override per-adapter. See docs/session-handoff.md."""
        return None

    @classmethod
    def terminal_resume_command(cls, sid: str) -> str | None:
        """Shell command that resumes session `sid` in a native terminal, or None if
        this agent can't be resumed by id from a terminal. Used by "Continue in
        Terminal" (headless cd's into the cwd first). Override per-adapter."""
        return None

    @classmethod
    async def list_models(cls, extra: dict | None = None) -> list[dict]:
        """Return [{id, label, detail}] for the new-session model picker.
        Override when provides_models=True. Called once at bridge startup before
        relay registration. extra = the DetectedAgent.extra dict (base_url, …).
        """
        return []


async def iter_json_lines(stream) -> "AsyncIterator[dict]":
    """Yield one parsed JSON object per line of an asyncio subprocess stream.

    The shared decode→strip→``json.loads`` loop every streaming-CLI adapter runs
    over its subprocess stdout: blank lines and non-JSON noise (banners, partial
    writes) are skipped, decode is lenient (``errors="replace"``). Only well-formed
    JSON objects reach the caller.
    """
    async for raw in stream:
        line = raw.decode(errors="replace").strip()
        if not line:
            continue
        try:
            yield json.loads(line)
        except json.JSONDecodeError:
            continue


async def read_stderr_tail(proc: "asyncio.subprocess.Process") -> str:
    """Drain a finished subprocess's stderr into a stripped string ("" if none).

    Safe to call after ``proc.wait()`` — the pipe still holds buffered output.
    Used to surface a startup/auth failure that produced no protocol events.
    """
    if proc.stderr is None:
        return ""
    try:
        return (await proc.stderr.read()).decode(errors="replace").strip()
    except Exception:
        return ""


class PerTurnCLIAdapter(BaseAdapter):
    """Base for *leaseless* CLI adapters — a fresh subprocess per turn, streaming a
    JSON-lines event protocol on stdout (codex, cursor, droid, gemini, goose).

    Owns the plumbing every one of them re-implemented identically: the send
    queue, ``send``/``stream``/``stop``, and a template ``_run_turn`` that brackets
    the turn with audio hints, spawns the process, drives the ``iter_json_lines`` →
    ``_parse_event`` loop, and emits the fallback stderr/end tail when the agent
    exits without a terminal ``SessionEndBlock``.

    A concrete adapter supplies just the two turn-specific pieces:

        _build_argv(text, attachments) -> list[str]   full argv incl. the prompt
        _parse_event(event) -> AsyncIterator[Block]   protocol → blocks

    and may override the hooks: ``_pre_spawn`` (refresh sentinel MCP, …),
    ``_turn_env`` (extra subprocess env beyond ``agent_env``), ``_reset_turn_state``
    (clear per-turn accumulators), and ``_on_no_session_end`` (custom no-events
    tail — gemini's rate-limit detection). Set ``cli_name`` for the "Failed to
    start …" message; ``_capture_stderr=False`` (codex) routes stderr to
    ``/dev/null``; ``_close_stdin=True`` (codex) closes stdin so the CLI doesn't
    block reading it.
    """

    cli_name: str = "agent"          # shown in the "Failed to start {cli_name}" error
    _capture_stderr: bool = True     # False → stderr=DEVNULL (codex)
    _close_stdin: bool = False       # True → stdin=DEVNULL (codex appends stdin to the prompt)

    # Set in a subclass's start(); the send/stream plumbing lives here.
    _send_queue: "asyncio.Queue[tuple[str, list[Attachment]]]"

    async def send(self, text: str, attachments: "list[Attachment] | None" = None) -> None:
        await self._send_queue.put((text, attachments or []))

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

    async def stop(self) -> None:
        proc = getattr(self, "_proc", None)
        if proc is not None and proc.returncode is None:
            proc.terminate()
            await proc.wait()

    # ── Hooks (override as needed) ────────────────────────────────────────────
    def _pre_spawn(self) -> None:
        """Called just before spawning — e.g. refresh the sentinel MCP script."""

    def _turn_env(self) -> "dict[str, str]":
        """Extra env for the subprocess, merged into ``agent_env`` (default none)."""
        return {}

    def _reset_turn_state(self) -> None:
        """Clear per-turn accumulators after a successful spawn (default none)."""

    async def _build_argv(self, text: str,
                          attachments: "list[Attachment]") -> "list[str]":
        raise NotImplementedError

    def _parse_event(self, event: dict) -> "AsyncIterator[Block]":
        raise NotImplementedError

    # ── Template turn loop ────────────────────────────────────────────────────
    async def _run_turn(self, text: str,
                        attachments: "list[Attachment]") -> "AsyncIterator[Block]":
        import asyncio as _asyncio
        from ..blocks import AudioHintBlock, ErrorBlock, SessionEndBlock

        yield AudioHintBlock(state="agent_speaking")
        argv = await self._build_argv(text, attachments)
        self._pre_spawn()
        write_session_claude_md(self._cwd)
        try:
            self._proc = await _asyncio.create_subprocess_exec(
                *argv,
                stdin=_asyncio.subprocess.DEVNULL if self._close_stdin else None,
                stdout=_asyncio.subprocess.PIPE,
                stderr=(_asyncio.subprocess.PIPE if self._capture_stderr
                        else _asyncio.subprocess.DEVNULL),
                cwd=self._cwd,
                env=agent_env(self._turn_env(), session_id=self._beacon_session_id),
            )
        except Exception as e:
            yield ErrorBlock(message=f"Failed to start {self.cli_name}: {e}", recoverable=True)
            yield SessionEndBlock(reason="error")
            return

        self._reset_turn_state()
        self._turn_saw_event = False
        session_ended = False
        try:
            assert self._proc.stdout is not None
            async for event in iter_json_lines(self._proc.stdout):
                self._turn_saw_event = True
                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()

        if not session_ended:
            async for block in self._on_no_session_end():
                yield block

    async def _on_no_session_end(self) -> "AsyncIterator[Block]":
        """Fallback tail when the process exited without a terminal SessionEndBlock:
        surface any stderr as an error, else complete by exit code."""
        from ..blocks import AudioHintBlock, ErrorBlock, SessionEndBlock
        proc = self._proc
        assert proc is not None
        err_txt = await read_stderr_tail(proc) if self._capture_stderr else ""
        yield AudioHintBlock(state="agent_silent")
        if (proc.returncode or 0) != 0 or err_txt:
            yield ErrorBlock(message=err_txt or f"{self.cli_name} exited with an error",
                             recoverable=True)
            yield SessionEndBlock(reason="error")
        else:
            yield SessionEndBlock(reason="completed")


# (pattern, context_window_tokens) — first match wins; more-specific patterns first.
_CONTEXT_WINDOWS: list[tuple[str, int]] = [
    ("claude",            200_000),   # all Anthropic Claude generations
    ("gpt-4.1",         1_047_576),
    ("gpt-4o",            128_000),
    ("gpt-4-turbo",       128_000),
    ("gpt-4",             128_000),
    ("o4-mini",           200_000),
    ("o3-mini",           200_000),
    ("o3",                200_000),
    ("o1-mini",           128_000),
    ("o1",                200_000),
    ("gpt-oss",           128_000),
    ("gemini-1.5-pro",  2_000_000),
    ("gemini-1.5",      1_048_576),
    ("gemini",          1_048_576),   # gemini-2.x and later
    ("grok",              131_072),
    ("llama",             128_000),
    ("mistral",           128_000),
    ("mixtral",           128_000),
    ("deepseek",          163_840),
    ("qwen",              131_072),
]


def context_window_for_model(model: str) -> int:
    """Return context window size in tokens for a model ID string.

    Handles prefixes like 'openrouter/', 'google/', 'anthropic/', etc.
    Returns 0 for unknown models.
    """
    if not model:
        return 0
    normalized = model.lower()
    # Extended-context variants carry a "[1m]" / "-1m" marker (e.g. Claude
    # "claude-opus-4-8[1m]", which has a 1M-token window). Claude strips this from the
    # model it reports in the stream, so callers should pass the LAUNCH model here.
    if "claude" in normalized and ("[1m]" in normalized or "-1m" in normalized
                                   or normalized.endswith("1m")):
        return 1_000_000
    for prefix in ("openrouter/", "kilo/", "google/", "anthropic/", "openai/",
                   "x-ai/", "meta-llama/", "mistralai/", "deepseek/", "qwen/"):
        if normalized.startswith(prefix):
            normalized = normalized[len(prefix):]
    for pattern, limit in _CONTEXT_WINDOWS:
        if pattern in normalized:
            return limit
    return 0


def _format_pricing(prompt_per_tok: float, completion_per_tok: float,
                    ctx: int | None = None) -> str:
    """Format token pricing into a human-readable detail string."""
    parts: list[str] = []
    if prompt_per_tok > 0:
        parts.append(f"${prompt_per_tok * 1_000_000:.4g}/1M in")
    if completion_per_tok > 0:
        parts.append(f"${completion_per_tok * 1_000_000:.4g}/1M out")
    if ctx:
        parts.append(f"{ctx // 1_000_000}M ctx" if ctx >= 1_000_000 else f"{ctx // 1_000}k ctx")
    return " · ".join(parts)
