"""``ApiFrontend`` - ``/api/*``: conversations, SSE, sessions, usage, limits (§3.9).

Bearer scope ``api``; token and audit management need ``admin``. Every
write goes through ``core/conversations``; the frontend only shapes JSON.
``/api/events`` and ``/api/conversations/{id}/events`` replay the gateway
bus as SSE with the same keepalive the markdown frontend uses, so a proxy
never sees a silent socket.

Usage figures come from the ``usage`` table (one row per turn, API-price
``cost_usd`` and per-model ``model_usage`` from the SDK's
``ResultMessage``); subscription quotas come from ``rate_limits``
(``RateLimitEvent``). The quota covers the whole subscription, so
``/api/limits`` puts the gateway's own spend for the window next to it
for calibration by eye.
"""

from __future__ import annotations

import json
import logging
import secrets
from collections import Counter
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING, Any, cast

from fastapi import FastAPI, HTTPException, Request, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, StreamingResponse
from sqlmodel import col, select

from beaver_gateway.core import audit
from beaver_gateway.core.auth import VALID_SCOPES, hash_token
from beaver_gateway.core.conversations import SEEDS, implied_title
from beaver_gateway.core.kinds import Kind, as_kind
from beaver_gateway.frontends._auth import require_token
from beaver_gateway.frontends._sse import (
    KEEPALIVE,
    SSE_HEADERS,
    events_with_heartbeat,
    sse_pack,
)
from beaver_gateway.frontends._urls import frontend_url
from beaver_gateway.frontends.base import Frontend
from beaver_gateway.storage import (
    create_token,
    list_audit_records,
    list_tokens,
    revoke_token,
)
from beaver_gateway.storage.models import (
    Conversation,
    InjectQueueItem,
    RateLimit,
    Token,
    Usage,
)

if TYPE_CHECKING:
    from collections.abc import AsyncIterator, Iterable, Sequence
    from pathlib import Path

    from beaver_gateway.core.conversations import Conversations
    from beaver_gateway.frontends.base import GatewayRuntime

_log = logging.getLogger("beaver_gateway.frontends.api")

__all__ = ["ApiFrontend"]

SCOPE = "api"
ADMIN_SCOPE = "admin"
GROUPS = ("day", "agent", "conversation", "model")
WINDOWS: dict[str, timedelta] = {
    "five_hour": timedelta(hours=5),
    "seven_day": timedelta(days=7),
    "seven_day_opus": timedelta(days=7),
    "seven_day_sonnet": timedelta(days=7),
}
MEMORY_MAX_DEPTH = 12
MEMORY_MAX_FILE = 2_000_000
MEMORY_MAX_ENTRIES = 5000


class ApiFrontend(Frontend):
    name = "api"
    kinds = ("master", "branch", "deep", "job")
    path = "/api"

    def __init__(
        self,
        *,
        master_agent: str | None = None,
        branch_agent: str | None = None,
        deep_agent: str | None = None,
        job_agent: str | None = None,
        memory_root: Path | None = None,
    ) -> None:
        self.master_agent = master_agent
        self.branch_agent = branch_agent
        self.deep_agent = deep_agent
        self.job_agent = job_agent
        self.memory_root = memory_root.resolve() if memory_root is not None else None
        self._app: FastAPI | None = None

    def agent_for(self, kind: Kind) -> str | None:
        return {
            "master": self.master_agent,
            "branch": self.branch_agent,
            "deep": self.deep_agent,
            "job": self.job_agent,
        }.get(kind)

    def configure(self, runtime: GatewayRuntime) -> None:
        if runtime.conversations is None or runtime.bus is None:
            msg = "ApiFrontend needs runtime.conversations and runtime.bus"
            raise RuntimeError(msg)
        self._app = build_app(runtime, memory_root=self.memory_root)

    def app(self) -> FastAPI | None:
        return self._app


def build_app(runtime: GatewayRuntime, *, memory_root: Path | None = None) -> FastAPI:  # noqa: PLR0915
    app = FastAPI(title="beaver-gateway / API")
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=False,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    conversations = cast("Conversations", runtime.conversations)

    async def body_of(request: Request) -> dict[str, Any]:
        if not await request.body():
            return {}
        try:
            data = await request.json()
        except json.JSONDecodeError as exc:
            raise HTTPException(
                status.HTTP_400_BAD_REQUEST, f"invalid JSON: {exc}"
            ) from exc
        if not isinstance(data, dict):
            raise HTTPException(status.HTTP_400_BAD_REQUEST, "body must be an object")
        return data

    async def conv_of(public_id: str) -> Conversation:
        conv = await conversations.get(public_id)
        if conv is None:
            raise HTTPException(
                status.HTTP_404_NOT_FOUND, f"unknown conversation {public_id}"
            )
        return conv

    def text_of(data: dict[str, Any], key: str = "text") -> str:
        text = data.get(key)
        if not isinstance(text, str) or not text.strip():
            raise HTTPException(status.HTTP_400_BAD_REQUEST, f"missing `{key}`")
        return text

    def int_or_none(data: dict[str, Any], key: str) -> int | None:
        value = data.get(key)
        if value is None:
            return None
        if not isinstance(value, int) or value < 1:
            raise HTTPException(
                status.HTTP_400_BAD_REQUEST, f"`{key}` must be a positive int"
            )
        return value

    def query_int(request: Request, key: str, default: int) -> int:
        raw = request.query_params.get(key)
        return int(raw) if raw and raw.isdigit() else default

    @app.get("/healthz")
    async def healthz() -> dict[str, str]:
        return {"status": "ok"}

    @app.get("/agents")
    async def list_agents(request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        return {
            "agents": [
                {
                    "name": a.name,
                    "model": a.model,
                    "type": a.__class__.__name__,
                    "kinds": list(getattr(a, "kinds", ())),
                    "effort": getattr(getattr(a, "options", None), "effort", None),
                }
                for a in runtime.agents
            ],
            "frontends": [
                {
                    "name": fe.name or fe.__class__.__name__,
                    "type": fe.__class__.__name__,
                    "kinds": list(fe.kinds),
                    "default_agents": {
                        k: fe.agent_for(k) for k in fe.kinds if fe.agent_for(k)
                    },
                    "path": fe.path,
                    "url": frontend_url(request, runtime, fe),
                }
                for fe in runtime.frontends
            ],
            "mcps": [{"name": m.name, "kind": m.kind} for m in runtime.mcps],
        }

    @app.get("/conversations")
    async def list_conversations(request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        q = request.query_params
        rows = await conversations.find(
            status=q.get("status"),
            kind=q.get("kind"),
            limit=query_int(request, "limit", 200),
        )
        latest = await conversations.queue.latest(cast("int", r.id) for r in rows)
        firsts = await conversations.first_user_texts(
            cast("int", r.id) for r in rows if not r.title
        )
        return {
            "conversations": [
                {
                    **conversations.public(r),
                    "title": r.title or implied_title(firsts.get(cast("int", r.id))),
                    "last_item": _queue_item(latest.get(cast("int", r.id))),
                }
                for r in rows
            ]
        }

    @app.post("/conversations", status_code=status.HTTP_201_CREATED)
    async def create_conversation(request: Request) -> dict[str, Any]:
        token = await require_token(request, runtime, scope=SCOPE)
        data = await body_of(request)
        agent = data.get("agent")
        try:
            kind = as_kind(str(data.get("kind") or "deep"))
        except ValueError as exc:
            raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
        if kind == "fork":
            raise HTTPException(status.HTTP_400_BAD_REQUEST, "fork is internal")
        if agent is not None and (
            not isinstance(agent, str) or agent not in runtime.agents
        ):
            raise HTTPException(status.HTTP_400_BAD_REQUEST, "unknown `agent`")
        seed = str(data.get("seed") or "clean")
        if seed not in SEEDS:
            raise HTTPException(
                status.HTTP_400_BAD_REQUEST, f"seed must be one of {SEEDS}"
            )
        parent = await conv_of(str(data["parent"])) if data.get("parent") else None
        try:
            conv = await conversations.spawn(
                kind=kind,
                agent=agent,
                seed=seed,
                parent=parent,
                text=data.get("text"),
                title=data.get("title"),
                window=int_or_none(data, "window"),
                origin="api",
            )
        except (ValueError, LookupError) as exc:
            raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
        await audit.log(
            runtime,
            actor=f"token:{token}",
            kind="api_spawn",
            agent_name=conv.agent_name,
            conversation=conv.external_id,
            seed=seed,
        )
        return await conversations.describe(conv)

    @app.get("/conversations/{public_id}")
    async def get_conversation(public_id: str, request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        conv = await conv_of(public_id)
        out = await conversations.describe(conv)
        out["queue"] = [
            _queue_item(i)
            for i in await conversations.queue.recent(cast("int", conv.id), limit=20)
        ]
        return out

    @app.get("/conversations/{public_id}/messages")
    async def get_messages(public_id: str, request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        conv = await conv_of(public_id)
        raw = request.query_params.get("window")
        window = int(raw) if raw and raw.isdigit() else None
        return {
            "id": conv.external_id,
            "text": await conversations.read(conv, window=window),
        }

    @app.get("/conversations/{public_id}/history")
    async def get_history(public_id: str, request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        conv = await conv_of(public_id)
        return {"id": conv.external_id, "messages": await conversations.history(conv)}

    @app.get("/conversations/{public_id}/entries")
    async def get_entries(public_id: str, request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        conv = await conv_of(public_id)
        subpath = request.query_params.get("subpath") or ""
        entries = await conversations.entries(conv, subpath=subpath)
        limit = query_int(request, "limit", 100)
        offset = query_int(request, "offset", max(len(entries) - limit, 0))
        return {
            "id": conv.external_id,
            "subpath": subpath,
            "subpaths": await conversations.subpaths(conv),
            "total": len(entries),
            "offset": offset,
            "entries": entries[offset : offset + limit],
        }

    @app.post(
        "/conversations/{public_id}/messages", status_code=status.HTTP_202_ACCEPTED
    )
    async def post_message(public_id: str, request: Request) -> dict[str, Any]:
        token = await require_token(request, runtime, scope=SCOPE)
        conv = await conv_of(public_id)
        data = await body_of(request)
        item = await conversations.post(
            conv, text_of(data), origin=str(data.get("origin") or "user")
        )
        await audit.log(
            runtime,
            actor=f"token:{token}",
            kind="api_message",
            agent_name=conv.agent_name,
            conversation=conv.external_id,
        )
        return {"id": conv.external_id, "item": item.id, "status": item.status}

    @app.post("/conversations/{public_id}/inject", status_code=status.HTTP_202_ACCEPTED)
    async def post_inject(public_id: str, request: Request) -> dict[str, Any]:
        token = await require_token(request, runtime, scope=SCOPE)
        conv = await conv_of(public_id)
        data = await body_of(request)
        urgency = str(data.get("urgency") or "normal")
        if urgency not in ("normal", "urgent"):
            raise HTTPException(
                status.HTTP_400_BAD_REQUEST, "urgency must be normal|urgent"
            )
        item = await conversations.inject(
            conv,
            text_of(data),
            urgency=cast("Any", urgency),
            origin=str(data.get("origin") or "api"),
        )
        await audit.log(
            runtime,
            actor=f"token:{token}",
            kind="api_inject",
            agent_name=conv.agent_name,
            conversation=conv.external_id,
            urgency=urgency,
        )
        return {"id": conv.external_id, "item": item.id, "priority": item.priority}

    @app.post("/conversations/{public_id}/say")
    async def post_say(public_id: str, request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        conv = await conv_of(public_id)
        return await conversations.say(conv, text_of(await body_of(request)))

    @app.post("/conversations/{public_id}/answer")
    async def post_answer(public_id: str, request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        conv = await conv_of(public_id)
        data = await body_of(request)
        question_id = text_of(data, "question_id")
        if not conversations.answer(question_id, text_of(data, "answer")):
            raise HTTPException(
                status.HTTP_404_NOT_FOUND, f"no open question {question_id}"
            )
        return {"id": conv.external_id, "question_id": question_id}

    @app.post("/conversations/{public_id}/branch", status_code=status.HTTP_201_CREATED)
    async def post_branch(public_id: str, request: Request) -> dict[str, Any]:
        token = await require_token(request, runtime, scope=SCOPE)
        parent = await conv_of(public_id)
        data = await body_of(request)
        seed = str(data.get("seed_mode") or data.get("seed") or "morning")
        if seed not in SEEDS:
            raise HTTPException(
                status.HTTP_400_BAD_REQUEST, f"seed must be one of {SEEDS}"
            )
        agent = data.get("agent")
        if agent is not None and not isinstance(agent, str):
            raise HTTPException(status.HTTP_400_BAD_REQUEST, "`agent` must be a string")
        try:
            child = await conversations.spawn(
                kind="branch",
                agent=agent,
                seed=seed,
                parent=parent,
                text=data.get("text"),
                title=data.get("title"),
                window=int_or_none(data, "window"),
                origin="api",
            )
        except (ValueError, LookupError) as exc:
            raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
        await audit.log(
            runtime,
            actor=f"token:{token}",
            kind="api_branch",
            agent_name=child.agent_name,
            conversation=child.external_id,
            parent=parent.external_id,
            seed=seed,
        )
        return await conversations.describe(child)

    @app.post("/conversations/{public_id}/merge")
    async def post_merge(public_id: str, request: Request) -> dict[str, Any]:
        token = await require_token(request, runtime, scope=SCOPE)
        conv = await conv_of(public_id)
        try:
            result = await conversations.merge(conv)
        except (ValueError, LookupError) as exc:
            raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
        await audit.log(
            runtime,
            actor=f"token:{token}",
            kind="api_merge",
            agent_name=conv.agent_name,
            conversation=conv.external_id,
        )
        return {
            "id": conv.external_id,
            "status": "merged",
            "fork": result.conversation.external_id,
            "text": result.text,
        }

    @app.post("/conversations/{public_id}/fork")
    async def post_fork(public_id: str, request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        conv = await conv_of(public_id)
        data = await body_of(request)
        try:
            result = await conversations.fork(
                conv,
                text_of(data, "prompt"),
                window=int_or_none(data, "window"),
                strip_tools=bool(data.get("strip_tools", False)),
            )
        except (ValueError, RuntimeError) as exc:
            raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
        return {
            "id": conv.external_id,
            "fork": result.conversation.external_id,
            "text": result.text,
        }

    @app.post("/conversations/{public_id}/bind")
    async def post_bind(public_id: str, request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        conv = await conv_of(public_id)
        data = await body_of(request)
        frontend = text_of(data, "frontend")
        external_id = text_of(data, "external_id")
        try:
            await conversations.bind(
                conv,
                frontend=frontend,
                external_id=external_id,
                visible=bool(data.get("visible", True)),
            )
        except ValueError as exc:
            raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
        return await conversations.describe(conv)

    @app.patch("/conversations/{public_id}/flags")
    async def patch_flags(public_id: str, request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        conv = await conv_of(public_id)
        data = await body_of(request)
        return conversations.public(await conversations.set_flags(conv, data))

    @app.patch("/conversations/{public_id}")
    async def patch_conversation(public_id: str, request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        conv = await conv_of(public_id)
        data = await body_of(request)
        try:
            if isinstance(data.get("status"), str):
                conv = await conversations.set_status(conv, data["status"])
            if isinstance(data.get("title"), str):
                conv = await conversations.set_title(conv, data["title"])
        except ValueError as exc:
            raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
        return conversations.public(conv)

    @app.get("/conversations/{public_id}/events")
    async def conversation_events(public_id: str, request: Request) -> Any:
        await require_token(request, runtime, scope=SCOPE)
        conv = await conv_of(public_id)
        return _sse(runtime, conversation_id=conv.external_id)

    @app.get("/events")
    async def all_events(request: Request) -> Any:
        await require_token(request, runtime, scope=SCOPE)
        return _sse(runtime, conversation_id=None)

    @app.get("/sessions")
    async def sessions(request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        pool = runtime.pool
        return {
            "rss": pool.rss() if pool is not None else None,
            "rss_limit": pool.rss_limit if pool is not None else None,
            "sessions": pool.snapshot() if pool is not None else [],
        }

    @app.get("/schedules")
    async def schedules(request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        raw = request.query_params.get("conversation")
        conv = await conv_of(raw) if raw else None
        return {
            "schedules": [
                {
                    "id": s.id,
                    "conversation_row": s.conversation_id,
                    "execute_at": _iso(s.execute_at),
                    "text": s.text,
                    "created_at": _iso(s.created_at),
                    "delivered_at": _iso(s.delivered_at),
                }
                for s in await conversations.schedules(conv)
            ]
        }

    @app.get("/usage")
    async def usage(request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        q = request.query_params
        group_by = q.get("group_by", "agent")
        if group_by not in GROUPS:
            raise HTTPException(
                status.HTTP_400_BAD_REQUEST, f"group_by must be one of {GROUPS}"
            )
        until = _parse_time(q.get("until")) or datetime.now(UTC)
        since = _parse_time(q.get("since")) or until - timedelta(
            hours=float(q.get("hours", "24"))
        )
        rows = await _usage_rows(runtime, since, until)
        groups = _group_usage(rows, group_by)
        if group_by == "conversation":
            titles = await _conversation_titles(runtime, [g["key"] for g in groups])
            for g in groups:
                g.update(titles.get(g["key"], {}))
        return {
            "since": since.isoformat(timespec="seconds"),
            "until": until.isoformat(timespec="seconds"),
            "group_by": group_by,
            "total": _sum_usage(rows),
            "rows": groups,
        }

    @app.get("/limits")
    async def limits(request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        rows = await conversations.rate_limits(limit=200)
        latest: dict[str, RateLimit] = {}
        for row in rows:
            latest.setdefault(row.window, row)
        now = datetime.now(UTC)
        windows = []
        for window, row in latest.items():
            length = WINDOWS.get(window)
            since = (
                _aware(row.resets_at) - length
                if length is not None and row.resets_at is not None
                else _aware(row.ts)
            )
            gateway = _sum_usage(await _usage_rows(runtime, since, now))
            gateway["since"] = since.isoformat(timespec="seconds")
            windows.append({**_limit_public(row), "gateway": gateway})
        windows.sort(
            key=lambda w: (WINDOWS.get(w["window"], timedelta.max), w["window"])
        )
        return {"windows": windows, "history": [_limit_public(r) for r in rows[:50]]}

    @app.get("/memory")
    async def memory_tree(request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        root = _memory_root(memory_root)
        return {"root": str(root), "tree": _tree(root, root, depth=0)}

    @app.get("/memory/file")
    async def memory_file(request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=SCOPE)
        root = _memory_root(memory_root)
        target = _memory_path(root, request.query_params.get("path") or "")
        if not target.is_file():
            raise HTTPException(status.HTTP_404_NOT_FOUND, "no such file")
        stat = target.stat()
        if stat.st_size > MEMORY_MAX_FILE:
            raise HTTPException(status.HTTP_413_CONTENT_TOO_LARGE, "file too large")
        try:
            content = target.read_text(encoding="utf-8")
        except UnicodeDecodeError as exc:
            raise HTTPException(
                status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, "not a text file"
            ) from exc
        return {
            "path": str(target.relative_to(root)),
            "size": stat.st_size,
            "mtime": _mtime(stat.st_mtime),
            "content": content,
        }

    @app.get("/tokens")
    async def tokens(request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=ADMIN_SCOPE)
        include_revoked = request.query_params.get("include_revoked") == "1"
        async with runtime.db.session() as session:
            rows = await list_tokens(session, include_revoked=include_revoked)
        return {"tokens": [_token_public(t) for t in rows]}

    @app.post("/tokens", status_code=status.HTTP_201_CREATED)
    async def token_create(request: Request) -> dict[str, Any]:
        actor = await require_token(request, runtime, scope=ADMIN_SCOPE)
        data = await body_of(request)
        name = text_of(data, "name").strip()
        scope = str(data.get("scope") or "*")
        if scope not in VALID_SCOPES:
            raise HTTPException(status.HTTP_400_BAD_REQUEST, f"invalid scope {scope!r}")
        plaintext = secrets.token_urlsafe(32)
        async with runtime.db.session() as session:
            try:
                row = await create_token(
                    session, name=name, scope=scope, hashed_value=hash_token(plaintext)
                )
            except Exception as exc:
                raise HTTPException(
                    status.HTTP_409_CONFLICT, f"could not create token {name!r}: {exc}"
                ) from exc
        await runtime.token_store.invalidate()
        await audit.log(
            runtime,
            actor=f"token:{actor}",
            kind="token_create",
            name=name,
            scope=scope,
            token_id=row.id,
        )
        return {"token": _token_public(row), "plaintext": plaintext}

    @app.post("/tokens/{token_id}/revoke")
    async def token_revoke(token_id: int, request: Request) -> dict[str, Any]:
        actor = await require_token(request, runtime, scope=ADMIN_SCOPE)
        async with runtime.db.session() as session:
            ok = await revoke_token(session, token_id=token_id)
        if not ok:
            raise HTTPException(
                status.HTTP_404_NOT_FOUND, f"no active token with id {token_id}"
            )
        await runtime.token_store.invalidate()
        await audit.log(
            runtime, actor=f"token:{actor}", kind="token_revoke", token_id=token_id
        )
        return {"id": token_id, "revoked": True}

    @app.get("/audit")
    async def audit_list(request: Request) -> dict[str, Any]:
        await require_token(request, runtime, scope=ADMIN_SCOPE)
        before_raw = request.query_params.get("before")
        before = int(before_raw) if before_raw and before_raw.isdigit() else None
        limit = min(query_int(request, "limit", 50), 500)
        async with runtime.db.session() as session:
            rows = await list_audit_records(session, limit=limit, before_id=before)
        return {
            "records": [
                {
                    "id": r.id,
                    "ts": _iso(r.ts),
                    "actor": r.actor,
                    "kind": r.kind,
                    "agent": r.agent_name,
                    "detail": _detail(r.detail_json),
                }
                for r in rows
            ],
            "next_before": rows[-1].id if len(rows) == limit else None,
        }

    @app.exception_handler(HTTPException)
    async def http_error(_request: Request, exc: HTTPException) -> JSONResponse:
        return JSONResponse(
            status_code=exc.status_code,
            content={"error": exc.detail},
            headers=exc.headers,
        )

    return app


def _sse(runtime: GatewayRuntime, *, conversation_id: str | None) -> StreamingResponse:
    async def gen() -> AsyncIterator[bytes]:
        stream = runtime.bus.stream(conversation_id=conversation_id)
        yield sse_pack("hello", {"type": "hello", "conversation_id": conversation_id})
        async for event in events_with_heartbeat(stream):
            if event is None:
                yield KEEPALIVE
                continue
            yield sse_pack(str(event["type"]), event)

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


def _parse_time(raw: str | None) -> datetime | None:
    if not raw:
        return None
    try:
        value = datetime.fromisoformat(raw)
    except ValueError as exc:
        raise HTTPException(
            status.HTTP_400_BAD_REQUEST, f"bad timestamp {raw!r}"
        ) from exc
    return _aware(value)


def _aware(value: datetime) -> datetime:
    return value if value.tzinfo is not None else value.replace(tzinfo=UTC)


def _iso(value: datetime | None) -> str | None:
    return _aware(value).isoformat(timespec="seconds") if value is not None else None


def _mtime(value: float) -> str:
    return datetime.fromtimestamp(value, tz=UTC).isoformat(timespec="seconds")


async def _usage_rows(
    runtime: GatewayRuntime, since: datetime, until: datetime
) -> list[Usage]:
    stmt = (
        select(Usage)
        .where(
            col(Usage.ts) >= since.astimezone(UTC).replace(tzinfo=None),
            col(Usage.ts) < until.astimezone(UTC).replace(tzinfo=None),
        )
        .order_by(col(Usage.ts))
    )
    async with runtime.db.session() as session:
        return list((await session.exec(stmt)).all())


def _empty() -> dict[str, Any]:
    return {
        "turns": 0,
        "input": 0,
        "output": 0,
        "cache_read": 0,
        "cache_creation": 0,
        "cost_usd": 0.0,
        "web_searches": 0,
    }


def _add(acc: dict[str, Any], row: Usage) -> None:
    acc["turns"] += 1
    acc["input"] += row.input_tokens
    acc["output"] += row.output_tokens
    acc["cache_read"] += row.cache_read_tokens
    acc["cache_creation"] += row.cache_creation_tokens
    acc["cost_usd"] += row.cost_usd or 0.0
    for per_model in (row.model_usage or {}).values():
        acc["web_searches"] += int(per_model.get("webSearchRequests") or 0)


def _add_model(acc: dict[str, Any], per_model: dict[str, Any]) -> None:
    acc["turns"] += 1
    acc["input"] += int(per_model.get("inputTokens") or 0)
    acc["output"] += int(per_model.get("outputTokens") or 0)
    acc["cache_read"] += int(per_model.get("cacheReadInputTokens") or 0)
    acc["cache_creation"] += int(per_model.get("cacheCreationInputTokens") or 0)
    acc["cost_usd"] += float(per_model.get("costUSD") or 0.0)
    acc["web_searches"] += int(per_model.get("webSearchRequests") or 0)


def _finish(acc: dict[str, Any]) -> dict[str, Any]:
    acc["cost_usd"] = round(acc["cost_usd"], 4)
    return acc


def _sum_usage(rows: Iterable[Usage]) -> dict[str, Any]:
    acc = _empty()
    for row in rows:
        _add(acc, row)
    return _finish(acc)


def _group_usage(rows: Sequence[Usage], group_by: str) -> list[dict[str, Any]]:
    groups: dict[str, dict[str, Any]] = {}
    agents: dict[str, Counter[str]] = {}
    for row in rows:
        if group_by == "model":
            per_model = row.model_usage or {}
            if not per_model:
                _add(groups.setdefault(row.model, _empty()), row)
            for model, mu in per_model.items():
                _add_model(groups.setdefault(model, _empty()), mu)
            continue
        key = _group_key(row, group_by)
        _add(groups.setdefault(key, _empty()), row)
        agents.setdefault(key, Counter())[row.agent_name] += 1
    out = []
    for key, acc in groups.items():
        entry = {"key": key, group_by: key, **_finish(acc)}
        if key in agents:
            entry["agent"] = agents[key].most_common(1)[0][0]
        out.append(entry)
    if group_by == "day":
        out.sort(key=lambda g: g["key"])
    else:
        out.sort(key=lambda g: (-g["cost_usd"], -g["output"], g["key"]))
    return out


def _group_key(row: Usage, group_by: str) -> str:
    if group_by == "day":
        return _aware(row.ts).date().isoformat()
    if group_by == "agent":
        return row.agent_name
    return row.conversation_id or "-"


async def _conversation_titles(
    runtime: GatewayRuntime, ids: Sequence[str]
) -> dict[str, dict[str, Any]]:
    wanted = [i for i in ids if i != "-"]
    if not wanted:
        return {}
    async with runtime.db.session() as session:
        rows = (
            await session.exec(
                select(Conversation).where(col(Conversation.external_id).in_(wanted))
            )
        ).all()
    return {
        r.external_id: {"title": r.title, "kind": r.kind, "status": r.status}
        for r in rows
    }


def _queue_item(item: InjectQueueItem | None) -> dict[str, Any] | None:
    if item is None:
        return None
    return {
        "id": item.id,
        "priority": item.priority,
        "origin": item.origin,
        "status": item.status,
        "created_at": _iso(item.created_at),
        "text": item.text[:200],
    }


def _limit_public(row: RateLimit) -> dict[str, Any]:
    return {
        "id": row.id,
        "ts": _iso(row.ts),
        "window": row.window,
        "status": row.status,
        "utilization": row.utilization,
        "resets_at": _iso(row.resets_at),
        "overage_status": row.overage_status,
        "overage_resets_at": _iso(row.overage_resets_at),
        "agent": row.agent_name,
        "session_id": row.session_id,
    }


def _token_public(row: Token) -> dict[str, Any]:
    return {
        "id": row.id,
        "name": row.name,
        "scope": row.scope,
        "created_at": _iso(row.created_at),
        "last_used_at": _iso(row.last_used_at),
        "revoked_at": _iso(row.revoked_at),
    }


def _detail(raw: str) -> Any:
    try:
        return json.loads(raw) if raw else {}
    except json.JSONDecodeError:
        return raw


def _memory_root(root: Path | None) -> Path:
    if root is None or not root.is_dir():
        raise HTTPException(
            status.HTTP_404_NOT_FOUND,
            "memory root is not configured (ApiFrontend(memory_root=...))",
        )
    return root.resolve()


def _memory_path(root: Path, raw: str) -> Path:
    target = (root / raw).resolve()
    if not target.is_relative_to(root):
        raise HTTPException(status.HTTP_404_NOT_FOUND, "no such file")
    return target


def _tree(root: Path, directory: Path, *, depth: int) -> list[dict[str, Any]]:
    if depth > MEMORY_MAX_DEPTH:
        return []
    out: list[dict[str, Any]] = []
    try:
        children = sorted(
            directory.iterdir(), key=lambda p: (not p.is_dir(), p.name.lower())
        )
    except OSError:
        return out
    for child in children:
        if child.name.startswith(".") or len(out) >= MEMORY_MAX_ENTRIES:
            continue
        try:
            stat = child.stat()
        except OSError:
            continue
        node: dict[str, Any] = {
            "name": child.name,
            "path": str(child.relative_to(root)),
            "type": "dir" if child.is_dir() else "file",
            "mtime": _mtime(stat.st_mtime),
        }
        if child.is_dir():
            node["children"] = _tree(root, child, depth=depth + 1)
        else:
            node["size"] = stat.st_size
        out.append(node)
    return out