"""The streamable-HTTP transport binding — the only module that imports ``mcp``.

One daemon, one protocol (docs/V3-PLAN.md §1.7): the MCP SDK's server core over
streamable HTTP on a fixed port, mounted at ``/mcp``. Local seats hit
``http://127.0.0.1:<port>/mcp``; remote seats reach the same server through
Tailscale Serve TLS (``https://<host>.<tailnet>.ts.net/mcp``). A Bearer header
identifies the seat on every request (:mod:`.seats`); requests with no valid
seat get 401 before they touch the protocol layer.

Tools come from the explicit registry (:mod:`.tools`) — the SDK server here is
a thin adapter (list/call), so schemas, seat filtering, and dispatch behavior
are identical with or without a transport in front (that's what the tests
exercise on Python versions the ``mcp`` extra doesn't support).
"""
import contextlib
import json
from contextvars import ContextVar

from .tools import ToolError, call_tool, list_tools

# The seat for the request being served, set by the auth middleware. Tasks the
# session manager spawns inherit a copy of the context, so tool handlers see it.
CURRENT_SEAT = ContextVar("vault_mcp_seat", default=None)

SERVER_NAME = "alexandria-vault"
MOUNT_PATH = "/mcp"
INBOUND_PATH = "/inbound-email"


def _instructions(svc) -> str:
    """The ≤2KB pointer (docs/DEVAULT.md step 5) — never the manual. Computed once
    at Server construction, so after the settings interview flips
    ``agent.personalized`` this text reads stale until the daemon restarts — both
    branches point clients at the live tools (capabilities/onboarding) for state."""
    agent_cfg = svc.config.get("agent") or {}
    agent = agent_cfg.get("address") or agent_cfg.get("name") or "the agent"
    if not agent_cfg.get("personalized", True):
        return (
            "This is a freshly installed Alexandria vault — not yet personalized: no "
            "owner name, and agent.address holds the placeholder 'Alexandria' (the "
            "SOFTWARE's name — never introduce yourself by it, and never offer it as "
            "a default or fallback; the interview replaces it). Brains running "
            "Alexandria have an ADDRESS, not a name — the owner has the name. On "
            "first contact, greet the owner ADDRESS-NEUTRALLY: open with 'This agent "
            "is running Alexandria v<version>' (the version is in capabilities()), "
            "say you don't have an address yet — they'll give you one — explain this "
            "vault becomes their Agent's Brain, and offer the ~10-minute setup "
            "interview. Once they say yes, the "
            "interview is DRIVEN: call interview_next(), ask the owner exactly the "
            "one question it returns as your whole message, and pass their reply to "
            "interview_answer() — which does the work and returns the next question. "
            "Do not compose your own interview from the settings playbook. "
            "capabilities() carries the personalized flag live. "
            "Every write commits to the vault's git "
            "history. These instructions are computed at server start — after the "
            "interview flips agent.personalized they may read stale until the server "
            "restarts; trust the tools, not this text.")
    return (f"This is an Alexandria vault — {agent}'s brain. Tools are the verbs: "
            "start with capabilities() (your seat + the tool map), and call manual() "
            "for the operating guide — the persona, filing conventions, and PEAK "
            "structure — before substantive work. Orient with vault_tree(). "
            "This brain ships named procedures: when a request sounds like a whole "
            "job rather than one tool call, skills() lists them and skill_read(name) "
            "returns the one that fits — read it instead of improvising the steps. "
            "Results are progressive — previews/summaries with cursors first; ask "
            "for detail='full' only when needed. Writes commit to the vault's git "
            "history; agentic tools return a job_id to poll with job_result. For "
            "multi-note synthesis questions, call think_deep rather than "
            "exploring manually. To tune the agent (name, voice, areas, digest), "
            "the onboarding tool serves the settings playbook.")


def build_mcp_server(svc):
    """The SDK server over the registry (requires the ``mcp`` extra)."""
    from mcp.server.lowlevel import Server
    from mcp import types

    server = Server(SERVER_NAME, instructions=_instructions(svc))

    @server.list_tools()
    async def _list_tools():
        # readOnlyHint carries the registry's scope over the wire so seats can
        # gate confirmation/plan-mode on the manifest instead of a local list.
        return [types.Tool(name=t["name"], description=t["description"],
                           inputSchema=t["inputSchema"],
                           annotations=types.ToolAnnotations(
                               readOnlyHint=(t.get("scope") != "write")))
                for t in list_tools(svc, CURRENT_SEAT.get())]

    @server.call_tool()
    async def _call_tool(name, arguments):
        seat = CURRENT_SEAT.get()
        if seat is None:  # transport auth should make this unreachable — fail closed
            raise ToolError("no authenticated seat on this request")
        result = call_tool(svc, seat, name, arguments or {})
        return [types.TextContent(type="text",
                                  text=json.dumps(result, ensure_ascii=False, default=str))]

    return server


class BearerAuthMiddleware:
    """Pure-ASGI Bearer check: resolve the seat or answer 401 — the protocol
    layer never sees an unauthenticated request."""

    def __init__(self, app, svc):
        self.app = app
        self.svc = svc

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return
        # The inbound-email webhook authenticates with its own shared secret (the
        # email provider can't hold a seat token), so it bypasses seat auth — the
        # route handler validates EMAIL_WEBHOOK_SECRET itself.
        if scope.get("path") == INBOUND_PATH:
            await self.app(scope, receive, send)
            return
        headers = {k.decode("latin-1").lower(): v.decode("latin-1")
                   for k, v in scope.get("headers") or []}
        auth = headers.get("authorization", "")
        token = auth[7:].strip() if auth.lower().startswith("bearer ") else ""
        seat = self.svc.seat_for_token(token)
        if seat is None:
            body = json.dumps({"error": "unauthorized: missing or unknown Bearer token"})
            await send({"type": "http.response.start", "status": 401,
                        "headers": [(b"content-type", b"application/json"),
                                    (b"www-authenticate", b"Bearer")]})
            await send({"type": "http.response.body", "body": body.encode("utf-8")})
            return
        reset = CURRENT_SEAT.set(seat)
        try:
            await self.app(scope, receive, send)
        finally:
            CURRENT_SEAT.reset(reset)


def build_asgi_app(svc):
    """The full ASGI app: Bearer auth → streamable-HTTP session manager at /mcp."""
    from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
    from starlette.applications import Starlette
    from starlette.routing import Mount, Route
    from starlette.responses import JSONResponse
    from starlette.concurrency import run_in_threadpool
    from . import inbound as inbound_mod

    manager = StreamableHTTPSessionManager(app=build_mcp_server(svc),
                                           json_response=True, stateless=True)

    async def handle(scope, receive, send):
        await manager.handle_request(scope, receive, send)

    async def _inbound(request):
        # The shared email gateway POSTs a parsed message here with the webhook
        # secret (Authorization: Bearer <secret> or X-Webhook-Secret). The model
        # ingest is blocking, so run it off the event loop.
        secret = request.headers.get("x-webhook-secret") or ""
        auth = request.headers.get("authorization", "")
        if not secret and auth.lower().startswith("bearer "):
            secret = auth[7:].strip()
        try:
            payload = await request.json()
        except Exception:  # noqa: BLE001 — malformed body → empty payload → gated out
            payload = {}
        result = await run_in_threadpool(inbound_mod.handle_inbound, svc, secret, payload)
        return JSONResponse(result, status_code=result.get("status", 200))

    @contextlib.asynccontextmanager
    async def lifespan(app):
        async with manager.run():
            yield

    inner = Starlette(routes=[
        Route(INBOUND_PATH, endpoint=_inbound, methods=["POST"]),
        Mount(MOUNT_PATH, app=handle),
    ], lifespan=lifespan)
    return BearerAuthMiddleware(inner, svc)


def bind_socket(host, port):
    """A bound, LISTENING TCP socket (``port=0`` = kernel-assigned ephemeral).

    Split out of :func:`serve` so the bind semantics are testable without
    uvicorn/the SDK: the sidecar contract is that the socket listens *before*
    the ready announce fires, so a parent that connects on sight of the
    announce can never race the bind — early requests queue in the accept
    backlog until the app is up, they don't fail."""
    import socket
    if ":" in host:  # IPv6 literal (e.g. `::` on IPv6-only private nets — Railway):
        # create_server defaults to AF_INET and refuses v6 addresses. Dual-stack
        # where the platform allows it, so 127.0.0.1 healthchecks keep working.
        try:
            return socket.create_server((host, port), family=socket.AF_INET6,
                                        dualstack_ipv6=True)
        except (ValueError, OSError):  # platform without dual-stack support
            return socket.create_server((host, port), family=socket.AF_INET6)
    return socket.create_server((host, port))



def timestamped_log_config():
    """uvicorn's default logging config with an ISO timestamp on every line.

    The stock formatters carry no date, so a launchd log of a long-lived
    server (`~/Library/Logs/vault-mcp.log`) cannot date a restart or a burst
    of 401s after the fact. Same formatters, same handlers — only ``fmt``
    gains a leading ``%(asctime)s`` and a ``datefmt`` (local time with
    offset, like the vault's sync commits)."""
    import copy
    from uvicorn.config import LOGGING_CONFIG
    cfg = copy.deepcopy(LOGGING_CONFIG)
    for fmt in cfg["formatters"].values():
        fmt["fmt"] = "%(asctime)s " + fmt["fmt"]
        fmt["datefmt"] = "%Y-%m-%dT%H:%M:%S%z"
    return cfg

def serve(svc, host, port, ready=None):
    """Run until stopped (uvicorn ships with the ``mcp`` extra).

    ``ready(actual_port)``, when given, fires after the socket is bound and
    listening — with ``port=0`` that is the only way to learn the real port
    (the Orion desktop's sidecar manager reads it off the announce line)."""
    import uvicorn
    server = uvicorn.Server(uvicorn.Config(build_asgi_app(svc), host=host,
                                           port=port, log_level="info",
                                           log_config=timestamped_log_config()))
    sock = bind_socket(host, port)
    if ready is not None:
        ready(sock.getsockname()[1])
    server.run(sockets=[sock])
