"""Loopback-only Hermes tool bridge with per-run, exact-scope grants."""

from __future__ import annotations

from dataclasses import dataclass, replace
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import hmac
import json
import secrets
import threading
import time
from typing import Any, Callable, Mapping

from .node import ACTION_POLICY
from .security import GrantAuthority, PeerIdentity


BRIDGE_PATH = "/v1/hermes/node-actions"
_CONTEXT_PREFIX = "OMNI_CONTEXT_V1 "


class BridgeError(ValueError):
    def __init__(self, status: int, code: str, detail: str) -> None:
        super().__init__(detail)
        self.status = status
        self.code = code
        self.detail = detail


@dataclass(frozen=True, slots=True)
class RunTicket:
    ticket: str
    session_id: str
    peer: PeerIdentity
    permissions: frozenset[str]
    expires_at: float
    run_id: str | None = None


class _BridgeHTTPServer(ThreadingHTTPServer):
    daemon_threads = True
    allow_reuse_address = True


class HermesToolBridge:
    """Issues no reusable grant; each accepted tool call gets one exact scope."""

    def __init__(
        self,
        *,
        enabled: bool,
        token: str,
        port: int,
        node_id: str,
        node_name: str,
        grant_authority: GrantAuthority | None,
        action_handler: Callable[[Mapping[str, Any], PeerIdentity], tuple[int, dict[str, object]]],
        peer_is_trusted: Callable[[str], bool],
        max_ticket_seconds: float = 3600.0,
    ) -> None:
        self.enabled = enabled
        self._token = token
        self._configured_port = port
        self.node_id = node_id
        self.node_name = node_name
        self._grants = grant_authority
        self._action_handler = action_handler
        self._peer_is_trusted = peer_is_trusted
        self._max_ticket_seconds = min(3600.0, max(60.0, max_ticket_seconds))
        self._tickets: dict[str, RunTicket] = {}
        self._lock = threading.RLock()
        self._server: _BridgeHTTPServer | None = None
        self._thread: threading.Thread | None = None

    @property
    def port(self) -> int:
        if self._server is not None:
            return int(self._server.server_address[1])
        return self._configured_port

    @property
    def endpoint(self) -> str:
        return f"http://127.0.0.1:{self.port}{BRIDGE_PATH}"

    @property
    def running(self) -> bool:
        return bool(self._thread and self._thread.is_alive())

    def status(self) -> dict[str, object]:
        with self._lock:
            active = sum(ticket.expires_at > time.time() for ticket in self._tickets.values())
        return {
            "enabled": self.enabled,
            "running": self.running,
            "bind": f"127.0.0.1:{self.port}" if self.enabled else None,
            "active_run_tickets": active,
            "authentication": "separate-bearer-and-run-ticket",
        }

    def start(self) -> None:
        if not self.enabled or self.running:
            return
        if len(self._token.encode("utf-8")) < 32 or self._grants is None:
            raise RuntimeError("Hermes bridge security is not configured")
        server = _BridgeHTTPServer(("127.0.0.1", self._configured_port), self._handler_class())
        self._server = server
        self._thread = threading.Thread(
            target=server.serve_forever,
            kwargs={"poll_interval": 0.1},
            name="omni-hermes-tool-bridge",
            daemon=True,
        )
        self._thread.start()

    def stop(self) -> None:
        self.revoke_all()
        server, self._server = self._server, None
        if server is not None:
            server.shutdown()
            server.server_close()
        thread, self._thread = self._thread, None
        if thread and thread.is_alive() and thread is not threading.current_thread():
            thread.join(timeout=1.0)

    def prepare_run(
        self,
        body: Mapping[str, Any],
        *,
        peer: PeerIdentity,
    ) -> tuple[dict[str, Any], str]:
        if not self.enabled or not self.running:
            raise BridgeError(503, "hermes_tool_bridge_unavailable", "local tool bridge is unavailable")
        session_id = str(body.get("session_id") or "")
        if not session_id or len(session_id) > 160:
            raise BridgeError(400, "invalid_session", "Hermes run requires a bounded session_id")
        context = _parse_context(body.get("instructions"))
        target = str(context.get("target_node") or "")
        if target and target.casefold() not in {
            self.node_id.casefold(),
            self.node_name.casefold(),
        }:
            raise BridgeError(409, "target_node_mismatch", "Hermes run targets another node")
        raw_permissions = context.get("requested_permissions", [])
        if not isinstance(raw_permissions, list) or any(
            not isinstance(permission, str) for permission in raw_permissions
        ):
            raise BridgeError(400, "invalid_permissions", "requested permissions are invalid")
        permissions = frozenset(raw_permissions)
        if not permissions.issubset(peer.roles):
            raise BridgeError(403, "permission_not_paired", "run requests an unpaired permission")
        ticket_value = secrets.token_urlsafe(32)
        ticket = RunTicket(
            ticket=ticket_value,
            session_id=session_id,
            peer=peer,
            permissions=permissions,
            expires_at=time.time() + self._max_ticket_seconds,
        )
        with self._lock:
            self._purge_expired_locked()
            self._tickets[ticket_value] = ticket
        context["node_tool_bridge"] = {
            "protocol_version": "1.0",
            "endpoint": self.endpoint,
            "run_ticket": ticket_value,
            "auth": "OMNI_HERMES_BRIDGE_TOKEN",
            "raw_media": False,
        }
        updated = dict(body)
        updated["instructions"] = _CONTEXT_PREFIX + json.dumps(
            context,
            ensure_ascii=True,
            sort_keys=True,
            separators=(",", ":"),
        )
        if len(updated["instructions"]) > 8192:
            self.revoke_ticket(ticket_value)
            raise BridgeError(413, "delegated_context_too_large", "delegated context is too large")
        return updated, ticket_value

    def bind_run(self, ticket_value: str, response_payload: bytes | None) -> str | None:
        run_id = None
        if response_payload:
            try:
                value = json.loads(response_payload)
                candidate = value.get("run_id") or value.get("id") if isinstance(value, Mapping) else None
                if isinstance(candidate, str) and 1 <= len(candidate) <= 160:
                    run_id = candidate
            except (UnicodeDecodeError, json.JSONDecodeError):
                run_id = None
        with self._lock:
            ticket = self._tickets.get(ticket_value)
            if ticket is None:
                return None
            if run_id is None:
                self._tickets.pop(ticket_value, None)
                return None
            self._tickets[ticket_value] = replace(ticket, run_id=run_id)
        return run_id

    def revoke_ticket(self, ticket_value: str) -> None:
        with self._lock:
            self._tickets.pop(ticket_value, None)

    def revoke_run(self, run_id: str) -> None:
        with self._lock:
            self._tickets = {
                value: ticket
                for value, ticket in self._tickets.items()
                if ticket.run_id != run_id
            }

    def revoke_peer(self, fingerprint: str) -> None:
        with self._lock:
            self._tickets = {
                value: ticket
                for value, ticket in self._tickets.items()
                if ticket.peer.certificate_fingerprint_sha256 != fingerprint
            }

    def revoke_all(self) -> None:
        with self._lock:
            self._tickets.clear()

    def perform(self, body: Mapping[str, Any]) -> tuple[int, dict[str, object]]:
        ticket_value = str(body.get("run_ticket") or "")
        with self._lock:
            self._purge_expired_locked()
            ticket = self._tickets.get(ticket_value)
        if ticket is None:
            raise BridgeError(403, "run_ticket_invalid", "run ticket is invalid or expired")
        if ticket.run_id is None:
            raise BridgeError(409, "run_ticket_not_bound", "Hermes run has not been accepted yet")
        if not self._peer_is_trusted(ticket.peer.certificate_fingerprint_sha256):
            self.revoke_ticket(ticket_value)
            raise BridgeError(403, "paired_peer_revoked", "paired peer is no longer trusted")
        action = body.get("action")
        if not isinstance(action, Mapping):
            raise BridgeError(400, "invalid_action", "action must be an object")
        action_type = str(action.get("type") or "")
        policy = ACTION_POLICY.get(action_type)
        if policy is None:
            raise BridgeError(400, "unsupported_action", "action type is not bridgeable")
        capability, permission = policy
        if str(action.get("permission") or "") != permission or permission not in ticket.permissions:
            raise BridgeError(403, "run_permission_denied", "run has no permission for this action")
        arguments = action.get("arguments", {})
        if not isinstance(arguments, Mapping):
            raise BridgeError(400, "invalid_arguments", "action arguments must be an object")
        action_id = str(action.get("action_id") or action.get("id") or "")
        if not action_id:
            action_id = f"action_bridge_{secrets.token_hex(10)}"
        operation_id = str(body.get("operation_id") or f"op_bridge_{secrets.token_hex(10)}")
        trace_id = str(body.get("trace_id") or f"trace_bridge_{secrets.token_hex(10)}")
        assert self._grants is not None
        now = time.time()
        grant = self._grants.sign(
            {
                "grant_id": f"grant_bridge_{secrets.token_hex(12)}",
                "node_id": self.node_id,
                "capability": capability,
                "session_id": ticket.session_id,
                "path_or_device_scope": _exact_scope(capability, arguments),
                "issued_at": now,
                "expires_at": min(ticket.expires_at, now + 60.0),
            }
        )
        request = {
            "node_id": self.node_id,
            "session_id": ticket.session_id,
            "operation_id": operation_id,
            "trace_id": trace_id,
            "action": {
                "action_id": action_id,
                "type": action_type,
                "permission": permission,
                "arguments": dict(arguments),
            },
            "grant": grant,
        }
        try:
            return self._action_handler(request, ticket.peer)
        except Exception as exc:
            status = getattr(exc, "status", None)
            code = getattr(exc, "code", None)
            detail = getattr(exc, "detail", None)
            if isinstance(status, int) and isinstance(code, str) and isinstance(detail, str):
                raise BridgeError(status, code, detail) from None
            raise

    def _purge_expired_locked(self) -> None:
        now = time.time()
        self._tickets = {
            value: ticket for value, ticket in self._tickets.items() if ticket.expires_at > now
        }

    def _handler_class(self) -> type[BaseHTTPRequestHandler]:
        bridge = self

        class Handler(BaseHTTPRequestHandler):
            server_version = "OMNIHermesBridge/1"
            sys_version = ""

            def log_message(self, _format: str, *_args: object) -> None:
                return None

            def do_GET(self) -> None:  # noqa: N802
                if not self._authorized():
                    return
                if self.path == "/health":
                    self._json(200, bridge.status())
                else:
                    self._json(404, {"error": "not_found"})

            def do_POST(self) -> None:  # noqa: N802
                if not self._authorized():
                    return
                if self.path != BRIDGE_PATH:
                    self._json(404, {"error": "not_found"})
                    return
                try:
                    body = self._read_object()
                    status, result = bridge.perform(body)
                except BridgeError as exc:
                    self._json(exc.status, {"error": exc.code, "detail": exc.detail})
                else:
                    self._json(status, result)

            def _authorized(self) -> bool:
                scheme, separator, value = self.headers.get("Authorization", "").partition(" ")
                valid = (
                    bool(separator)
                    and scheme.lower() == "bearer"
                    and hmac.compare_digest(value.strip().encode(), bridge._token.encode())
                )
                if not valid:
                    self._json(401, {"error": "unauthorized"})
                return valid

            def _read_object(self) -> dict[str, Any]:
                if self.headers.get("Content-Type", "").split(";", 1)[0].lower() != "application/json":
                    raise BridgeError(415, "unsupported_media_type", "use application/json")
                try:
                    length = int(self.headers.get("Content-Length", ""))
                except ValueError as exc:
                    raise BridgeError(411, "content_length_required", "valid length required") from exc
                if not 0 <= length <= 64 * 1024:
                    raise BridgeError(413, "payload_too_large", "bridge request is too large")
                try:
                    value = json.loads(self.rfile.read(length))
                except (UnicodeDecodeError, json.JSONDecodeError) as exc:
                    raise BridgeError(400, "invalid_json", "malformed JSON") from exc
                if not isinstance(value, dict):
                    raise BridgeError(400, "invalid_body", "JSON root must be an object")
                return value

            def _json(self, status: int, value: Mapping[str, object]) -> None:
                payload = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode()
                self.send_response(status)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(payload)))
                self.send_header("Cache-Control", "no-store")
                self.end_headers()
                self.wfile.write(payload)

        return Handler


def _parse_context(instructions: object) -> dict[str, Any]:
    text = str(instructions or "")
    if not text.startswith(_CONTEXT_PREFIX):
        return {}
    try:
        value = json.loads(text[len(_CONTEXT_PREFIX) :])
    except json.JSONDecodeError as exc:
        raise BridgeError(400, "invalid_omni_context", "OMNI context is malformed") from exc
    if not isinstance(value, dict):
        raise BridgeError(400, "invalid_omni_context", "OMNI context must be an object")
    return value


def _exact_scope(capability: str, arguments: Mapping[str, Any]) -> dict[str, object]:
    if capability in {"files.read", "files.write"}:
        path = arguments.get("path") or arguments.get("source")
        return {"path_prefixes": [str(path or "")]}
    if capability == "webcam.capture":
        return {"device_ids": [str(arguments.get("device_id") or "")]}
    if capability == "screen.capture":
        return {
            "screen_ids": [
                "all"
                if arguments.get("all_screens") is True
                else str(arguments.get("screen_id") or "primary")
            ]
        }
    return {}