"""Process entrypoint.

Phase 1.4 — async ``main``: install uvloop, load the user config, build
registries + per-agent backends (only ``RaycastBackend`` so far), wire
each frontend with a ``GatewayRuntime``, and run all
``frontend.serve()`` coroutines concurrently. Without any frontends we
still print the Phase 0 DoD line and exit cleanly so the bare skeleton
keeps working.

Phase 2.1 — when the user declares any ``McpServer``, we additionally
build the internal MCP aggregator app and run it on
``127.0.0.1:INTERNAL_MCP_PORT`` as another task inside the same
TaskGroup. URLs are surfaced through ``GatewayRuntime.mcp_internal_urls``
so Phase 2.2's ClaudeCode adapter can find them.

Phase 3 — the same aggregator backs the external ``McpServerFrontend``;
``cli`` doesn't have to know that the frontend reverse-proxies into it,
it just keeps the aggregator running for anyone who needs it.
"""

from __future__ import annotations

import asyncio
import contextlib
import functools
import logging
import signal
from contextlib import AsyncExitStack
from typing import TYPE_CHECKING, Any

import psycopg
import uvicorn
import uvloop
from dotenv import load_dotenv
from pgqueuer import PsycopgDriver
from raycast_api import Client as RaycastClient
from raycast_api.config import Config as RaycastConfig

from beaver_gateway import config_loader
from beaver_gateway.agents.claude import ClaudeAgent
from beaver_gateway.agents.raycast import RaycastAgent
from beaver_gateway.backends.claude_sdk import (
    ClaudeSdkBackend,
    RunnerConfig,
    UsageEvent,
)
from beaver_gateway.backends.raycast import RaycastBackend
from beaver_gateway.core.auth import TokenStore
from beaver_gateway.core.bus import EventBus
from beaver_gateway.core.conversations import Conversations
from beaver_gateway.core.envelope import Envelope
from beaver_gateway.core.gateway_tools import build_tool_server
from beaver_gateway.core.redact import install as install_redaction
from beaver_gateway.core.redact import load_secrets as load_secrets_to_mask
from beaver_gateway.core.registry import AgentRegistry, Gateway, McpRegistry
from beaver_gateway.core.rotation import Rotation, RotationPolicy
from beaver_gateway.core.scheduler import Scheduler
from beaver_gateway.core.sessions import SessionPool
from beaver_gateway.frontends._auth import require_token
from beaver_gateway.frontends.base import GatewayRuntime
from beaver_gateway.frontends.root import build_root_app
from beaver_gateway.mcp.internal_app import build_internal_app
from beaver_gateway.settings import Settings
from beaver_gateway.storage import (
    Database,
    PostgresSessionStore,
    Usage,
    append_audit,
    append_usage,
)

if TYPE_CHECKING:
    from claude_agent_sdk import McpSdkServerConfig
    from fastmcp import FastMCP
    from fastmcp.tools.base import Tool as FastMCPTool
    from starlette.applications import Starlette
    from starlette.requests import Request
    from starlette.types import ASGIApp

    from beaver_gateway.backends.base import Backend
    from beaver_gateway.core.policy import ToolAudit
    from beaver_gateway.mcp.types import McpServerT


_log = logging.getLogger("beaver_gateway.cli")


def main() -> None:
    """Sync wrapper: uvloop loop factory + asyncio.run."""
    logging.basicConfig(
        level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s"
    )
    install_redaction()
    _install_sigterm_handler()
    asyncio.run(_async_main(), loop_factory=uvloop.new_event_loop)


def _install_sigterm_handler() -> None:
    """Turn SIGTERM into a normal interpreter exit.

    Python's default SIGTERM disposition kills the process outright, so
    neither ``AsyncExitStack`` unwinding nor ``atexit`` hooks run. That
    matters because every ``claude`` we spawn lives in its own session
    (ptyprocess calls ``setsid``), which makes it immune to the signal
    that took us down — a hard SIGTERM leaves one orphaned CLI per live
    session, each holding hundreds of MB. Raising ``KeyboardInterrupt``
    instead routes ``docker stop`` / ``systemctl stop`` through the same
    shutdown path as Ctrl-C, which does reap them.

    Only installed when we own the main thread's signal handlers; under
    an embedding host that isn't ours to take.
    """

    def _raise_interrupt(_signum: int, _frame: object) -> None:
        raise KeyboardInterrupt

    with contextlib.suppress(ValueError, OSError):
        signal.signal(signal.SIGTERM, _raise_interrupt)


async def _async_main() -> None:
    # Populate ``os.environ`` from ``.env`` before anything else so the
    # user's ``config.py`` can read its own secrets via ``os.environ[...]``
    # (Firefly PAT, third-party MCP creds, etc.). ``Settings`` already
    # reads ``.env`` independently via pydantic-settings, but that path
    # populates Settings fields, not the process environment.
    # ``override=False``: real env vars (Docker, systemd) win over .env.
    load_dotenv(override=False)
    # Only now does the process environment hold the credentials the
    # redactor masks literally (in Docker they arrive via ``env_file``).
    load_secrets_to_mask()

    settings = Settings()  # ty: ignore[missing-argument]
    gateway = config_loader.load(settings.config_path)

    agents = AgentRegistry(gateway.agents)
    mcps = McpRegistry(gateway.mcps)

    # Phase 4.1 — open the async DB and run create_all once. Engine
    # pool is process-wide; ``dispose()`` after the TaskGroup unwinds.
    db = Database(settings.database_url)
    await db.create_all()

    # Phase 4.2 — TokenStore now reads from the DB (in-memory cache
    # primed at start, TTL-refreshed, last_used_at flushed by a
    # background task). BOOTSTRAP_TOKENS layers on top so first-run /
    # examples still work without DB writes.
    token_store = TokenStore(
        db,
        bootstrap=TokenStore.parse_bootstrap(settings.bootstrap_tokens),
        bootstrap_scopes=TokenStore.parse_bootstrap_scopes(settings.bootstrap_tokens),
    )

    async with AsyncExitStack() as stack:
        stack.push_async_callback(db.dispose)
        await token_store.start()
        stack.push_async_callback(token_store.stop)
        # Internal MCP URLs must exist before we construct any
        # ClaudeSdkBackend - adapters bake the URLs into their
        # ``mcp_servers`` at construction time. The
        # ``mcp_servers`` map is used by the Raycast backend, which
        # needs in-process ``list_tools`` / ``call_tool`` access (the
        # Raycast wire has no native MCP concept).
        internal_app, internal_urls, mcp_servers = _build_internal_mcp(
            gateway.mcps, settings=settings
        )

        # Prefetch tool catalogs for every MCP so RaycastAgent requests
        # don't pay a per-turn list_tools roundtrip and so a broken MCP
        # surfaces at startup instead of mid-conversation.
        mcp_tools = await _prefetch_mcp_tools(mcp_servers)

        pool = SessionPool()
        bus = EventBus()
        late = _LateConversations()
        session_store = PostgresSessionStore(db)
        backends: dict[str, Backend] = await _build_backends(
            settings=settings,
            agents=agents,
            stack=stack,
            db=db,
            session_store=session_store,
            mcp_internal_urls=internal_urls,
            mcp_servers=mcp_servers,
            mcp_tools=mcp_tools,
            pool=pool,
            late=late,
        )
        conversations = Conversations(
            db=db,
            agents=agents,
            backends=backends,
            bus=bus,
            pool=pool,
            store=session_store,
            texts=gateway.texts,
            frontends=gateway.frontends,
            envelope=Envelope(
                watch=gateway.watch, tz=gateway.tz, recall=gateway.recall
            ),
            distiller=gateway.distiller,
            user_sink=gateway.user_sink,
        )
        late.conversations = conversations
        scheduler = Scheduler(
            conversations=conversations,
            jobs=gateway.jobs,
            driver=await _pgqueuer_driver(settings.database_url, stack),
            budget=gateway.budget,
            rotation=Rotation(
                conversations, gateway.rotation or RotationPolicy(tz=gateway.tz)
            ),
            tz=gateway.tz,
        )
        conversations.scheduler = scheduler

        runtime = GatewayRuntime(
            agents=agents,
            mcps=mcps,
            backends=backends,
            token_store=token_store,
            db=db,
            mcp_internal_urls=internal_urls,
            admin_user=settings.admin_user,
            admin_pass=settings.admin_pass,
            session_secret=settings.session_secret,
            frontends=tuple(gateway.frontends),
            conversations=conversations,
            bus=bus,
            pool=pool,
            scheduler=scheduler,
            public_url=gateway.public_url.rstrip("/") if gateway.public_url else None,
        )

        for fe in gateway.frontends:
            fe.configure(runtime)

        _log.info(
            "beaver-gateway: loaded %d agents, %d mcps, %d frontends",
            len(agents),
            len(mcps),
            len(gateway.frontends),
        )

        # Keep the Phase 0 DoD line on stdout for grep-friendly smoke
        # tests, in addition to the structured log line above.
        print(
            f"beaver-gateway: loaded {len(agents)} agents, "
            f"{len(mcps)} mcps, {len(gateway.frontends)} frontends"
        )

        if not gateway.frontends:
            # No external listeners → nothing to serve. The internal
            # MCP app has no consumer on its own, so we skip running
            # it in this path and exit cleanly (Phase 0 DoD).
            return

        await conversations.start()
        stack.push_async_callback(conversations.stop)
        await scheduler.start()
        stack.push_async_callback(scheduler.stop)
        hooks = scheduler.app(
            functools.partial(_authorize_hook, runtime=runtime, scope="api")
        )
        async with asyncio.TaskGroup() as tg:
            tg.create_task(pool.reap_loop())
            if internal_app is not None:
                tg.create_task(_serve_internal_mcp(internal_app, settings=settings))
            tg.create_task(_serve_root(gateway, extra={"/hooks": hooks}))
            if gateway.watch is not None:
                tg.create_task(gateway.watch.run())
            for fe in gateway.frontends:
                tg.create_task(fe.serve())


async def _authorize_hook(
    request: Request, *, runtime: GatewayRuntime, scope: str
) -> str:
    return await require_token(request, runtime, scope=scope)


async def _pgqueuer_driver(url: str, stack: AsyncExitStack) -> PsycopgDriver | None:
    """A dedicated autocommit connection for pgqueuer's LISTEN/NOTIFY."""
    plain = _plain_postgres_url(url)
    if plain is None:
        return None
    conn = await psycopg.AsyncConnection.connect(plain, autocommit=True)
    stack.push_async_callback(conn.close)
    return PsycopgDriver(conn)


def _plain_postgres_url(url: str) -> str | None:
    for prefix in ("postgresql+psycopg://", "postgresql://", "postgres://"):
        if url.startswith(prefix):
            return "postgresql://" + url[len(prefix) :]
    return None


async def _serve_root(gateway: Gateway, *, extra: dict[str, ASGIApp]) -> None:
    app = build_root_app(gateway.frontends, extra=extra)
    config = uvicorn.Config(app, host=gateway.host, port=gateway.port, log_level="info")
    _log.info(
        "gateway on http://%s:%d - %s",
        gateway.host,
        gateway.port,
        ", ".join([*(fe.path for fe in gateway.frontends if fe.path), *extra])
        or "no http frontends",
    )
    await uvicorn.Server(config).serve()


def _build_internal_mcp(
    mcps: list[McpServerT], *, settings: Settings
) -> tuple[Starlette | None, dict[str, str], dict[str, FastMCP]]:
    """Build the aggregator app + URL map + server map, or empty equivalents.

    The URL map is always handed out (frontends may still introspect
    ``runtime.mcp_internal_urls`` even if nothing is configured); the
    app is ``None`` when there are no MCPs to mount, so the caller
    skips the uvicorn task entirely. The server map is the in-process
    handle the Raycast backend needs to splice MCP tools into its
    requests — empty when no MCPs are configured.
    """
    if not mcps:
        return None, {}, {}
    return build_internal_app(mcps, host="127.0.0.1", port=settings.internal_mcp_port)


async def _prefetch_mcp_tools(
    servers: dict[str, FastMCP],
) -> dict[str, list[FastMCPTool]]:
    """Eagerly enumerate tools per MCP so the Raycast loop has a static catalog.

    Each underlying proxy is allowed to fail independently — a broken
    MCP shouldn't take down the whole gateway. The result has one entry
    per MCP that responded; agents that ``expose_mcps`` a missing entry
    will simply expose no tools from it (logged once per request).
    """
    out: dict[str, list[FastMCPTool]] = {}
    for name, server in servers.items():
        try:
            out[name] = list(await server.list_tools())
        except Exception:  # noqa: BLE001 — proxy can raise any transport error; we degrade per-MCP rather than fail the whole gateway
            _log.exception("failed to list tools for MCP %r — skipping", name)
            out[name] = []
    return out


async def _serve_internal_mcp(app: Starlette, *, settings: Settings) -> None:
    """Run the internal MCP aggregator on loopback.

    Bound to ``127.0.0.1`` (never EXPOSE'd) — only the in-process
    ClaudeCode subprocess reaches it. Logged at ``warning`` level so
    we don't drown the gateway's own logs in per-request noise.
    """
    config = uvicorn.Config(
        app,
        host="127.0.0.1",
        port=settings.internal_mcp_port,
        log_level="warning",
        loop="uvloop",
    )
    server = uvicorn.Server(config)
    _log.info(
        "internal MCP aggregator on http://127.0.0.1:%d/mcp/<name>",
        settings.internal_mcp_port,
    )
    await server.serve()


class _LateConversations:
    """Backends need a tool-server factory before the service that backs it exists."""

    conversations: Conversations | None = None

    def server(
        self, key: str, _kind: str, names: tuple[str, ...]
    ) -> McpSdkServerConfig | None:
        if self.conversations is None or not names:
            return None
        return build_tool_server(self.conversations, conversation_key=key, names=names)

    async def ask(self, key: str, payload: dict[str, Any]) -> str:
        if self.conversations is None:
            msg = "conversations service is not up yet"
            raise RuntimeError(msg)
        answer = await self.conversations.ask(key, payload)
        return self.conversations.answer_text(answer)


async def _build_backends(
    *,
    settings: Settings,
    agents: AgentRegistry,
    stack: AsyncExitStack,
    db: Database,
    session_store: PostgresSessionStore,
    mcp_internal_urls: dict[str, str],
    mcp_servers: dict[str, FastMCP],
    mcp_tools: dict[str, list[FastMCPTool]],
    pool: SessionPool,
    late: _LateConversations,
) -> dict[str, Backend]:
    """Construct one backend per agent name.

    The Raycast ``Client`` is shared across every ``RaycastAgent``
    (bearer + device-id are process-wide), so we open it lazily - only
    when at least one ``RaycastAgent`` is present - and close it via
    the caller's exit stack.

    Each :class:`ClaudeAgent` gets its own :class:`ClaudeSdkBackend`
    (own live-session pool, own prompt and MCP set); all of them share
    the session store and the usage sink.
    """
    backends: dict[str, Backend] = {}

    raycast_agents = [a for a in agents if isinstance(a, RaycastAgent)]
    if raycast_agents:
        client = await _try_open_raycast_client(settings, stack)
        if client is not None:
            raycast_backend = RaycastBackend(
                client, mcp_servers=mcp_servers, mcp_tools=mcp_tools
            )
            for a in raycast_agents:
                backends[a.name] = raycast_backend

    runner = RunnerConfig(user=settings.claude_runner_user, home=settings.claude_home)
    mcp_tool_names = {
        name: [t.name for t in tools] for name, tools in mcp_tools.items()
    }

    async def record_usage(event: UsageEvent) -> None:
        row = Usage(
            agent_name=event.agent_name,
            conversation_id=event.conversation_id,
            session_id=event.session_id,
            model=event.model,
            effort=event.effort,
            input_tokens=event.usage.input_tokens,
            output_tokens=event.usage.output_tokens,
            cache_read_tokens=event.usage.cache_read_tokens,
            cache_creation_tokens=event.usage.cache_creation_tokens,
            context_tokens=event.usage.context_tokens,
            cost_usd=event.usage.cost_usd,
            duration_ms=event.usage.duration_ms,
            num_turns=event.usage.num_turns,
            model_usage=event.usage.model_usage,
        )
        try:
            async with db.session() as session:
                await append_usage(session, row)
        except Exception:  # noqa: BLE001
            _log.exception("usage write failed for %s", event.agent_name)

    async def record_tool(event: ToolAudit) -> None:
        detail = {
            "conversation": event.conversation,
            "kind": event.kind,
            "tool": event.tool,
            "decision": event.decision,
            "reason": event.reason,
            "brief": event.brief,
        }
        try:
            async with db.session() as session:
                await append_audit(
                    session,
                    actor=f"agent:{event.agent}",
                    kind="tool_call",
                    agent_name=event.agent,
                    detail=detail,
                )
        except Exception:  # noqa: BLE001
            _log.exception("tool audit write failed for %s", event.agent)

    for a in agents:
        if isinstance(a, ClaudeAgent):
            adapter = ClaudeSdkBackend(
                agent=a,
                mcp_internal_urls=mcp_internal_urls,
                session_store=session_store,
                mcp_tool_names=mcp_tool_names,
                runner=runner,
                usage_sink=record_usage,
                pool=pool,
                tool_server=functools.partial(late.server, names=a.gateway_tools),
                asker=late.ask,
                audit_sink=record_tool,
            )
            await stack.enter_async_context(adapter)
            backends[a.name] = adapter

    return backends


async def _try_open_raycast_client(
    settings: Settings, stack: AsyncExitStack
) -> RaycastClient | None:
    """Open a ``raycast_api.Client`` if creds are available, else warn + skip.

    Skipping is gentler than failing startup: the user can bring up the
    gateway, list their agents, and still hit a ``ClaudeAgent``; affected
    ``RaycastAgent`` instances 503 with a specific message at request time.
    Missing ``raycast.json`` falls into the same bucket — first-run
    users won't have it yet.
    """
    if not settings.raycast_bearer:
        _log.warning(
            "RaycastAgent present but RAYCAST_BEARER is unset — those agents will 503"
        )
        return None
    if not settings.raycast_device_id:
        _log.warning(
            "RaycastAgent present but RAYCAST_DEVICE_ID is unset — those agents "
            "will 503 (generate once with `python -c 'import secrets; "
            "print(secrets.token_hex(32))'`)"
        )
        return None
    if not settings.raycast_config_path.exists():
        _log.warning(
            "RaycastAgent present but %s is missing — those agents will 503",
            settings.raycast_config_path,
        )
        return None

    config = RaycastConfig.load(settings.raycast_config_path)
    client = RaycastClient(
        config=config,
        bearer_token=settings.raycast_bearer,
        device_id=settings.raycast_device_id,
        locale=settings.raycast_locale,
    )
    return await stack.enter_async_context(client)