"""Starlette app factory + endpoint handlers for ratatoskr.web.

Per docs/contracts/issues/16.contract.md. INV-002: create_app accepts
a client_factory callable; the factory produces a configured
httpx.AsyncClient. Tests pass a respx-mocked factory; production
passes a factory that bakes in WORLDTREE_API_URL + WORLDTREE_API_KEY.
"""

from __future__ import annotations

import asyncio
import itertools
import json
from collections.abc import AsyncIterator, Callable
from dataclasses import asdict, dataclass, is_dataclass
from importlib.metadata import version as _pkg_version

import httpx
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import (
    FileResponse,
    JSONResponse,
    Response,
    StreamingResponse,
)
from starlette.routing import Mount, Route
from starlette.staticfiles import StaticFiles

from ratatoskr import local_agents as _local_agents
from ratatoskr.first_message import seed_preset_first_message
from ratatoskr.sessions import (
    AgentNotAvailable,
    AgentNotFound,
    AuthScopeDenied,
    BifrostBinding,
    BifrostConsumerKeyMissing,
    BifrostHandshakeFailed,
    PersonaNotConfigured,
    SessionApiFailed,
    create_session,
    endpoint_for_plane,
    get_persona_state,
    get_session_bifrost,
    get_session_messages,
    get_session_tools,
    list_agents,
)
from ratatoskr.sse_client import (
    AdminEvent,
    CancelAlreadyCompleted,
    CancelFailed,
    Cancelled,
    CancelTurnNotFound,
    Done,
    Error,
    MalformedSseData,
    MalformedSseId,
    SseConnectFailed,
    SseConnectionDropped,
    TurnIdFlip,
    cancel_turn,
    stream_admin_events,
    stream_turn_resilient,
)


def _static_dir() -> str:
    """Locate the bundled static/ directory inside the installed package.

    Uses importlib.resources so the lookup works for editable installs,
    wheel installs, and uvicorn's worker reload. Per INV-009 packaging:
    static/index.html ships in the wheel.
    """
    from importlib.resources import files
    return str(files("ratatoskr.web") / "static")


def _root_endpoint(request: Request) -> FileResponse:
    """GET / → index.html. Per FN root_endpoint POST-001."""
    from pathlib import Path
    return FileResponse(
        Path(_static_dir()) / "index.html",
        media_type="text/html",
    )


def _version_endpoint(request: Request) -> JSONResponse:
    """GET /version → {"ratatoskr": "<version>"}.

    Per FN version_endpoint POST-001.
    """
    return JSONResponse({"ratatoskr": _pkg_version("ratatoskr")}, status_code=200)


def _as_dict(obj: object) -> dict:
    """Best-effort dataclass-to-dict for AgentInfo / LocalAgentEntry."""
    if is_dataclass(obj):
        return asdict(obj)
    return dict(obj)  # type: ignore[arg-type]


async def _agents_endpoint(request: Request) -> JSONResponse:
    """GET /api/agents → upstream /agents + local Tier 3 index merge.

    Per FN agents_endpoint POST-001 + ERRORS table.
    """
    client_factory = request.app.state.client_factory
    try:
        async with client_factory() as client:
            upstream = await list_agents(client)
    except SessionApiFailed as exc:
        return JSONResponse(
            {"error_code": "session_api_failed", "status": exc.status},
            status_code=exc.status,
        )
    except httpx.RequestError as exc:
        return JSONResponse(
            {"error_code": "network_error", "message": str(exc)},
            status_code=502,
        )
    upstream_ids = {a.agent_id for a in upstream}
    local = _local_agents.load_local_agents()
    merged = [_as_dict(a) for a in upstream] + [
        _as_dict(le) for le in local if le.agent_id not in upstream_ids
    ]
    return JSONResponse(merged, status_code=200)


async def _create_session_endpoint(request: Request) -> JSONResponse:
    """POST /api/sessions → upstream POST /sessions. Per FN create_session_endpoint.

    v0.16.0: end_user_id is SERVER-configured (app.state.end_user_id from
    RATATOSKR_END_USER_ID), never read from the browser body. A client
    cannot impersonate an arbitrary end-user partition.
    """
    body = await request.json()
    agent_id = body.get("agent_id") if isinstance(body, dict) else None
    if not agent_id:
        return JSONResponse({"error_code": "missing_agent_id"}, status_code=400)
    end_user_id = request.app.state.end_user_id
    client_factory = request.app.state.client_factory

    # Issue #17 (web bind split): the browser may select a PLANE; the server holds
    # the consumer key + visible host and constructs the binding. The consumer key
    # NEVER reaches the browser (INV-008/INV-009).
    bifrost: BifrostBinding | None = None
    bifrost_plane = body.get("bifrost_plane") if isinstance(body, dict) else None
    consumer_key = request.app.state.bifrost_consumer_key
    visible_host = request.app.state.bifrost_visible_host
    if bifrost_plane:
        if bifrost_plane not in ("memory", "affect", "combined"):
            return JSONResponse(
                {"error_code": "invalid_bifrost_plane"}, status_code=400
            )
        if not (consumer_key and visible_host):
            return JSONResponse(
                {"error_code": "bifrost_not_configured"}, status_code=400
            )
        bifrost = BifrostBinding(
            endpoint_url=endpoint_for_plane(bifrost_plane, visible_host)
        )

    try:
        async with client_factory() as client:
            info = await create_session(
                client,
                agent_id,
                end_user_id=end_user_id,
                bifrost=bifrost,
                consumer_key=consumer_key if bifrost else None,
            )
            # #347 authored first-message: seed the agent's preset opening
            # (best-effort; never blocks create — see first_message INV-001).
            await seed_preset_first_message(client, info.session_id, agent_id)
    except AgentNotFound:
        return JSONResponse({"error_code": "agent_not_found"}, status_code=404)
    except BifrostConsumerKeyMissing:
        # Server misconfiguration: a plane was requested but no consumer key.
        return JSONResponse(
            {"error_code": "bifrost_not_configured"}, status_code=400
        )
    except BifrostHandshakeFailed as exc:
        return JSONResponse(
            {
                "error_code": "bifrost_handshake_failed",
                "bifrost_error": exc.bifrost_error,
            },
            status_code=502,
        )
    except SessionApiFailed as exc:
        return JSONResponse(
            {"error_code": "session_api_failed", "status": exc.status},
            status_code=exc.status,
        )
    payload = _as_dict(info)
    if bifrost is not None:
        # Bound-state for the UI indicator — plane + endpoint only, never the key.
        payload["bifrost"] = {
            "plane": bifrost_plane,
            "endpoint": bifrost.endpoint_url,
            "status": "bound",
        }
    return JSONResponse(payload, status_code=201)


@dataclass
class TurnHandle:
    """In-flight turn record stored in app.state.turn_registry.

    Per FN submit_turn_endpoint + INV-005/006/007.

    v0.16.0: `upstream_turn_id` captures Worldtree's server-assigned
    turn_id (from the first SSE event's sse_id.turn_id) once the stream
    opens. Cancel paths target THIS, not the browser-local `turn_id` —
    the local counter is only a registry key. None until the first
    upstream event arrives; cancel before then is a no-op (nothing to
    cancel upstream yet).
    """

    session_id: str
    turn_id: int
    content: str
    status: str = "queued"  # queued | streaming | done | error | cancelled
    upstream_turn_id: int | None = None


# Process-local monotonic turn_id counter. Per FN submit_turn_endpoint
# STEPS 2: turn_id is opaque to the upstream Worldtree (whose own
# turn_ids come back via SSE); the registry's key uses our own counter
# so cancel/stream lookups don't need upstream-issued ids.
_TURN_COUNTER = itertools.count(1)


async def _submit_turn_endpoint(request: Request) -> JSONResponse:
    """POST /api/turns/{session_id} → allocate turn_id + register handle.

    Per FN submit_turn_endpoint. Does NOT open the upstream stream here;
    the subsequent GET /api/turns/{sid}/stream does that.
    """
    body = await request.json()
    content = body.get("content") if isinstance(body, dict) else None
    if not content:
        return JSONResponse({"error_code": "missing_content"}, status_code=400)
    session_id = request.path_params["session_id"]
    turn_id = next(_TURN_COUNTER)
    request.app.state.turn_registry[(session_id, turn_id)] = TurnHandle(
        session_id=session_id, turn_id=turn_id, content=content,
    )
    return JSONResponse({"turn_id": turn_id}, status_code=200)


def _event_to_browser_payload(event: object) -> tuple[str, dict]:
    """Serialize an upstream Event dataclass to (browser_event_type, json_dict).

    Per INV-008 + FN stream_turn_endpoint STEP 3. The dict shape is
    locked by tests/fixtures/presentation_contract.json — one entry per
    Event type. Implementation: snake_case class name as event_type;
    asdict(event) with sse_id flattened to "T:S" string.
    """
    type_name = type(event).__name__
    # CamelCase → snake_case
    browser_type = "".join(
        ("_" + c.lower() if c.isupper() and i else c.lower())
        for i, c in enumerate(type_name)
    )
    data = asdict(event)  # type: ignore[arg-type]
    sse_id = data.get("sse_id")
    if isinstance(sse_id, (list, tuple)) and len(sse_id) == 2:
        data["sse_id"] = f"{sse_id[0]}:{sse_id[1]}"
    elif isinstance(sse_id, dict) and "turn_id" in sse_id and "seq" in sse_id:
        data["sse_id"] = f"{sse_id['turn_id']}:{sse_id['seq']}"
    return browser_type, data


def _format_sse(event_type: str, data: dict) -> bytes:
    """Format a browser-facing SSE event with `event:` + `data:`.

    Two-newline terminator per the SSE spec.
    """
    return f"event: {event_type}\ndata: {json.dumps(data)}\n\n".encode()


async def _stream_turn_endpoint(request: Request) -> StreamingResponse:
    """GET /api/turns/{session_id}/stream?turn_id=N → proxy upstream SSE.

    Per FN stream_turn_endpoint. Handles browser-disconnect cleanup
    (INV-005) and synthesizes `event: error` for upstream typed
    exceptions.
    """
    session_id = request.path_params["session_id"]
    try:
        turn_id = int(request.query_params["turn_id"])
    except (KeyError, ValueError):
        return JSONResponse({"error_code": "missing_turn_id"}, status_code=400)
    registry = request.app.state.turn_registry
    handle = registry.get((session_id, turn_id))
    if handle is None:
        return JSONResponse({"error_code": "turn_not_found"}, status_code=404)

    client_factory = request.app.state.client_factory

    async def gen() -> AsyncIterator[bytes]:
        client = client_factory()
        try:
            handle.status = "streaming"
            try:
                async for event in stream_turn_resilient(client, session_id, handle.content):
                    # v0.16.0: capture the upstream (Worldtree-assigned)
                    # turn_id from the first event so cancel paths target
                    # the real upstream turn, not our local counter.
                    if handle.upstream_turn_id is None:
                        sse_id = getattr(event, "sse_id", None)
                        if sse_id is not None:
                            handle.upstream_turn_id = sse_id.turn_id
                    event_type, data = _event_to_browser_payload(event)
                    yield _format_sse(event_type, data)
                    if isinstance(event, (Done, Error, Cancelled)):
                        handle.status = type(event).__name__.lower()
                        break
            except (SseConnectFailed, SseConnectionDropped, MalformedSseId,
                    MalformedSseData, TurnIdFlip) as exc:
                yield _format_sse(
                    "error",
                    {"exception": type(exc).__name__, "message": str(exc)},
                )
                handle.status = "error"
            except asyncio.CancelledError:
                # Browser disconnect path (INV-005). Cancel the UPSTREAM
                # turn (if it started) — never the local turn_id.
                if handle.status == "streaming" and handle.upstream_turn_id is not None:
                    try:
                        await cancel_turn(client, session_id, handle.upstream_turn_id)
                    except (CancelAlreadyCompleted, CancelTurnNotFound):
                        pass  # cooperative race — turn already terminal upstream
                    except Exception as exc:
                        # v0.16.1: unexpected cancel failure during disconnect
                        # cleanup (e.g. CancelFailed, transport error) — log for
                        # diagnosability instead of silently swallowing. Never
                        # re-raise: we're already unwinding the cancelled
                        # generator and must not mask the CancelledError.
                        import sys as _sys
                        _sys.stderr.write(
                            f'{{"kind":"disconnect_cancel","event":"cancel_failed",'
                            f'"session_id":"{session_id}",'
                            f'"upstream_turn_id":{handle.upstream_turn_id},'
                            f'"exc":"{type(exc).__name__}"}}\n'
                        )
                raise
        finally:
            registry.pop((session_id, turn_id), None)
            await client.aclose()

    return StreamingResponse(gen(), media_type="text/event-stream")


async def _cancel_turn_endpoint(request: Request) -> JSONResponse:
    """POST /api/turns/{session_id}/cancel?turn_id=N. Per FN cancel_turn_endpoint."""
    session_id = request.path_params["session_id"]
    try:
        turn_id = int(request.query_params["turn_id"])
    except (KeyError, ValueError):
        return JSONResponse({"error_code": "missing_turn_id"}, status_code=400)
    registry = request.app.state.turn_registry
    handle = registry.get((session_id, turn_id))
    if handle is None:
        return JSONResponse({"error_code": "turn_not_found"}, status_code=404)
    # v0.16.0: cancel targets the UPSTREAM turn_id captured during
    # streaming, not the browser-local turn_id. If the upstream stream
    # never started (upstream_turn_id is None), there's nothing to
    # cancel — clean up and report a no-op.
    if handle.upstream_turn_id is None:
        registry.pop((session_id, turn_id), None)
        return JSONResponse(
            {"cancelled": False, "reason": "not_started"}, status_code=200
        )
    client_factory = request.app.state.client_factory
    try:
        async with client_factory() as client:
            await cancel_turn(client, session_id, handle.upstream_turn_id)
        body = {"cancelled": True}
    except (CancelAlreadyCompleted, CancelTurnNotFound):
        body = {"cancelled": False, "reason": "race_or_completed"}
    except CancelFailed as exc:
        registry.pop((session_id, turn_id), None)
        return JSONResponse(
            {"error_code": "cancel_failed", "status": exc.status},
            status_code=exc.status,
        )
    registry.pop((session_id, turn_id), None)
    return JSONResponse(body, status_code=200)


async def _persona_state_endpoint(request: Request) -> JSONResponse:
    """GET /api/agents/{agent_id}/persona_state. Per FN persona_state_endpoint."""
    agent_id = request.path_params["agent_id"]
    client_factory = request.app.state.client_factory
    try:
        async with client_factory() as client:
            snap = await get_persona_state(client, agent_id)
    except PersonaNotConfigured:
        return JSONResponse({"error_code": "persona_not_configured"}, status_code=404)
    except AgentNotAvailable:
        return JSONResponse({"error_code": "agent_not_available"}, status_code=404)
    except AuthScopeDenied:
        return JSONResponse({"error_code": "auth_scope_denied"}, status_code=403)
    return JSONResponse(snap, status_code=200)


async def _affect_state_endpoint(request: Request) -> JSONResponse:
    """GET /api/affect/{agent_id} → proxy the provider PAD read route. Supplies
    end_user_id SERVER-SIDE (never the browser, INV-002); proxies to the configured
    affect-read URL, re-encoding agent_id into the path (colon-id safe, INV-008).
    Per FN affect_state_endpoint (#18 Deliverable 2)."""
    from urllib.parse import quote

    agent_id = request.path_params["agent_id"]
    affect_read_url = request.app.state.affect_read_url
    end_user_id = request.app.state.end_user_id
    if not (affect_read_url and end_user_id):  # PRE-001: fail-visible, never silent
        return JSONResponse({"error_code": "affect_not_configured"}, status_code=400)
    url = f"{affect_read_url}/affect/state/{quote(agent_id, safe='')}"
    try:
        async with httpx.AsyncClient() as client:
            r = await client.get(url, params={"end_user_id": end_user_id})
    except httpx.RequestError:
        return JSONResponse(
            {"error_code": "affect_provider_unreachable"}, status_code=502
        )
    return JSONResponse(r.json(), status_code=r.status_code)


async def _session_tools_endpoint(request: Request) -> JSONResponse:
    """GET /api/sessions/{session_id}/tools → owner-scoped tool inventory (spec #183).

    Proxies get_session_tools with the client's CONSUMER bearer (no admin scope):
    the merged {agent_id, builtin_tools, bifrost_tools} the LLM saw at turn-fire.
    Any non-200 upstream → surfaced as a status-preserving error envelope."""
    session_id = request.path_params["session_id"]
    client_factory = request.app.state.client_factory
    try:
        async with client_factory() as client:
            info = await get_session_tools(client, session_id)
    except SessionApiFailed as exc:
        return JSONResponse(
            {"error_code": "session_tools_unavailable", "status": exc.status},
            status_code=exc.status,
        )
    return JSONResponse(info, status_code=200)


async def _session_messages_endpoint(request: Request) -> JSONResponse:
    """GET /api/sessions/{session_id}/messages → the session's message history.

    Proxies get_session_messages so the SPA can render a session's EXISTING turns
    on open — notably a #347 authored first-message seeded at create-time (which
    lives in the ledger, not the live turn stream). Any non-200 upstream → a
    status-preserving error envelope."""
    session_id = request.path_params["session_id"]
    client_factory = request.app.state.client_factory
    try:
        async with client_factory() as client:
            data = await get_session_messages(client, session_id)
    except SessionApiFailed as exc:
        return JSONResponse(
            {"error_code": "session_messages_unavailable", "status": exc.status},
            status_code=exc.status,
        )
    return JSONResponse(data, status_code=200)


async def _session_bifrost_endpoint(request: Request) -> JSONResponse:
    """GET /api/sessions/{session_id}/bifrost → admin-scoped Bifrost dispatch state (#176).

    The admin key is SERVER-HELD (app.state.admin_key) and never reaches the
    browser (INV-003 precedent — upstream credentials stay server-side); the
    wrapper overrides the Authorization header with it. Fail-visible when the
    admin key isn't configured (never a silent empty pane)."""
    session_id = request.path_params["session_id"]
    admin_key = request.app.state.admin_key
    if not admin_key:  # PRE-001: fail-visible, never silent
        return JSONResponse({"error_code": "admin_key_not_configured"}, status_code=400)
    client_factory = request.app.state.client_factory
    try:
        async with client_factory() as client:
            bstate = await get_session_bifrost(client, session_id, admin_key=admin_key)
    except SessionApiFailed as exc:
        return JSONResponse(
            {"error_code": "bifrost_state_unavailable", "status": exc.status},
            status_code=exc.status,
        )
    return JSONResponse(bstate, status_code=200)


def _admin_event_matches_web(ev: AdminEvent, session_id: str | None) -> bool:
    """AdminEvents filter (design-brief §6, mirrors the TUI): forward non-heartbeat
    system.* (stream-integrity signals) + events for the active session; drop the
    rest so the browser sees only session-relevant lifecycle, never the full
    cross-session admin firehose."""
    if ev.type == "system.heartbeat":
        return False
    if ev.type.startswith("system."):
        return True
    return session_id is not None and ev.data.get("session_id") == session_id


async def _admin_events_endpoint(request: Request) -> Response:
    """GET /api/admin/events?session_id=... → SSE proxy of GET /admin/events (#11).

    The admin key is SERVER-HELD; the browser only ever receives the session-filtered
    stream (never the key, never the cross-session firehose). Long-lived + best-effort:
    a connect failure or mid-stream drop emits a labeled `stream_error` event and ends."""
    admin_key = request.app.state.admin_key
    if not admin_key:  # PRE-001: fail-visible, never silent
        return JSONResponse({"error_code": "admin_key_not_configured"}, status_code=400)
    session_id = request.query_params.get("session_id")
    client_factory = request.app.state.client_factory

    async def gen() -> AsyncIterator[bytes]:
        client = client_factory()
        try:
            async for ev in stream_admin_events(client, admin_key=admin_key):
                if not _admin_event_matches_web(ev, session_id):
                    continue
                # Fixed SSE event name so the browser renders EVERY admin type
                # with one listener (no per-type enumeration → nothing silently
                # dropped); the real dotted type rides in the payload.
                yield _format_sse(
                    "admin_event",
                    {"id": ev.id, "type": ev.type, "timestamp": ev.timestamp,
                     "data": ev.data},
                )
        except (SseConnectFailed, SseConnectionDropped, MalformedSseId,
                MalformedSseData) as exc:
            yield _format_sse(
                "stream_error",
                {"exception": type(exc).__name__, "message": str(exc)},
            )
        except asyncio.CancelledError:
            raise  # browser disconnect — let the generator unwind
        finally:
            await client.aclose()

    return StreamingResponse(gen(), media_type="text/event-stream")


def create_app(
    client_factory: Callable[[], httpx.AsyncClient],
    *,
    end_user_id: str | None = None,
    bifrost_consumer_key: str | None = None,
    bifrost_visible_host: str | None = None,
    affect_read_url: str | None = None,
    admin_key: str | None = None,
) -> Starlette:
    """Construct the Starlette app — wire routes + state per FN create_app.

    INV-002: app MUST NOT construct httpx.AsyncClient at module top;
    everything HTTP-bound goes through client_factory.
    INV-006: lifespan shutdown drains the turn registry within a 5s
    budget — every in-flight turn gets a best-effort upstream cancel.

    v0.16.0: `end_user_id` is the server-configured Worldtree end-user
    partition (from RATATOSKR_END_USER_ID). Threaded into POST /sessions
    server-side; never accepted from the browser.
    """
    assert callable(client_factory)

    from contextlib import asynccontextmanager

    @asynccontextmanager
    async def lifespan(app: Starlette):
        yield
        # Shutdown path — drain in-flight turns per INV-006. Cancel the
        # UPSTREAM turn_id (v0.16.0); skip handles whose upstream stream
        # never started (upstream_turn_id is None — nothing to cancel).
        import sys as _sys

        registry: dict[tuple[str, int], TurnHandle] = app.state.turn_registry
        in_flight = [
            h for h in registry.values()
            if h.status == "streaming" and h.upstream_turn_id is not None
        ]
        if in_flight:
            client = client_factory()
            try:
                task_to_handle = {
                    asyncio.create_task(
                        cancel_turn(client, h.session_id, h.upstream_turn_id)
                    ): h
                    for h in in_flight
                }
                done, pending = await asyncio.wait(task_to_handle, timeout=5.0)
                # Per-pending session/turn detail (INV-006 logging fidelity).
                for task in pending:
                    h = task_to_handle[task]
                    task.cancel()
                    _sys.stderr.write(
                        f'{{"kind":"shutdown","event":"cleanup_timeout",'
                        f'"session_id":"{h.session_id}",'
                        f'"upstream_turn_id":{h.upstream_turn_id}}}\n'
                    )
            finally:
                await client.aclose()
        registry.clear()

    routes = [
        Route("/", _root_endpoint),
        Mount("/static", app=StaticFiles(directory=_static_dir()), name="static"),
        Route("/version", _version_endpoint),
        Route("/api/agents", _agents_endpoint),
        Route("/api/sessions", _create_session_endpoint, methods=["POST"]),
        Route("/api/agents/{agent_id}/persona_state", _persona_state_endpoint),
        Route("/api/affect/{agent_id}", _affect_state_endpoint),
        Route("/api/sessions/{session_id}/tools", _session_tools_endpoint),
        Route("/api/sessions/{session_id}/messages", _session_messages_endpoint),
        Route("/api/sessions/{session_id}/bifrost", _session_bifrost_endpoint),
        Route("/api/admin/events", _admin_events_endpoint),
        Route("/api/turns/{session_id}", _submit_turn_endpoint, methods=["POST"]),
        Route("/api/turns/{session_id}/stream", _stream_turn_endpoint),
        Route("/api/turns/{session_id}/cancel", _cancel_turn_endpoint, methods=["POST"]),
    ]
    app = Starlette(routes=routes, lifespan=lifespan)
    app.state.client_factory = client_factory
    app.state.end_user_id = end_user_id
    # Issue #17 (web bind split): the consumer key + Worldtree-visible provider
    # host are SERVER-HELD config (env), never sent from the browser. The browser
    # selects only the PLANE; the server constructs the bound session (INV-008).
    app.state.bifrost_consumer_key = bifrost_consumer_key
    app.state.bifrost_visible_host = bifrost_visible_host
    # Issue #18 (Deliverable 2): the provider affect-read base URL (server→provider hop,
    # same dev box) — distinct from the WT-visible host used for binding.
    app.state.affect_read_url = affect_read_url
    # Admin observability panes (BifrostState + AdminEvents): the admin key is
    # SERVER-HELD (RATATOSKR_ADMIN_API_KEY) and never reaches the browser — the
    # server proxies admin-scoped reads and forwards only the session-filtered result.
    app.state.admin_key = admin_key
    # INV-002: turn registry is in-process memory, keyed (session_id, turn_id)
    app.state.turn_registry = {}
    return app
