"""Loopback approval server for the MCP Proxy approval surface."""

from __future__ import annotations

from dataclasses import dataclass
from html import escape
import hashlib
import json
import hmac
import os
import shutil
import signal
import subprocess
from http import HTTPStatus
from pathlib import Path
from http.cookies import SimpleCookie
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import secrets
import threading
import time
from typing import Any, Callable, Mapping
from urllib.parse import parse_qs

from agentveil_mcp_proxy.evidence.events_show import (
    LOCAL_PROOF_AGENT_PROMPT,
    LOCAL_PROOF_BLOCK_TITLE,
    LOCAL_PROOF_PENDING_QUIET_LINE,
    LOCAL_PROOF_POST_DENY_BODY,
)
from agentveil_mcp_proxy.evidence.observability import (
    approval_proof_detail_rows,
    approval_raw_evidence_rows,
    bounded_action_display,
    bounded_reason_for_record,
    bounded_resource_display,
    human_approval_reason_label,
    human_approval_summary,
    pending_approval_dict,
    risk_class_plain_label,
    terminal_state_for_record_status,
)


MAX_POST_BODY_BYTES = 8192
REQUEST_SOCKET_TIMEOUT_SECONDS = 5.0
DEFAULT_TERMINAL_REQUEST_RETENTION_SECONDS = 600.0
MIN_TERMINAL_REQUEST_RETENTION_SECONDS = 1.0
SECURITY_HEADERS = {
    "Referrer-Policy": "no-referrer",
    "Cache-Control": "no-store, no-cache, must-revalidate, max-age=0",
    "Pragma": "no-cache",
    "Expires": "0",
    "X-Frame-Options": "DENY",
    "Content-Security-Policy": (
        "default-src 'self'; script-src 'self'; "
        "style-src 'self' 'unsafe-inline'; frame-ancestors 'none'"
    ),
    "X-Content-Type-Options": "nosniff",
}
COOKIE_NAME = "avp_approval_session"
INTERNAL_REGISTER_TOKEN_HEADER = "X-AVP-Approval-Register-Token"
APPROVAL_LOCAL_URL_WARNING = (
    "This local approval URL is a bearer-style session URL. Do not share it."
)
APPROVAL_DECISION_RECORDED_BODY = (
    "Decision recorded. Retry the same MCP tool call without changing tool, target, or payload."
)
APPROVAL_POST_APPROVE_HUMAN_BODY = (
    "Approval recorded. Return to your agent. "
    "AgentVeil is ready for the approved action to continue."
)
APPROVAL_POST_APPROVE_PROOF_BODY = "To inspect what happened, ask your agent:"
APPROVAL_POST_APPROVE_PROOF_PROMPT = "Show AgentVeil local proof for the last action."
TERMINAL_ALREADY_APPROVED_BODY = "This request was already approved."
TERMINAL_ALREADY_APPROVED_NEXT = (
    "Return to your agent. AgentVeil is ready for the approved action to continue."
)
TERMINAL_ALREADY_DENIED_BODY = "This request was already denied."
TERMINAL_EXPIRED_BODY = "This approval request has expired."
TERMINAL_NO_LONGER_ACTIONABLE_BODY = "This approval request is no longer actionable."
APPROVAL_DECISION_DENIED_BODY = (
    "Decision recorded. This action was denied and will not run."
)


LOCAL_PROOF_COPY_COMMAND_LABEL = "Copy prompt"
_LOCAL_PROOF_COMMAND_ELEMENT_ID = "approval-local-proof-command"


def generate_approval_script_nonce() -> str:
    """Return a per-response nonce for Approval Center inline scripts."""

    return secrets.token_urlsafe(16)


def approval_content_security_policy(*, script_nonce: str | None = None) -> str:
    """Build Approval Center CSP, optionally allowing one inline script nonce."""

    script_src = "script-src 'self'"
    if script_nonce:
        script_src = f"script-src 'self' 'nonce-{script_nonce}'"
    return (
        f"default-src 'self'; {script_src}; "
        "style-src 'self' 'unsafe-inline'; frame-ancestors 'none'"
    )


def approval_security_headers(*, script_nonce: str | None = None) -> dict[str, str]:
    """Return security headers for one Approval Center response."""

    headers = dict(SECURITY_HEADERS)
    headers["Content-Security-Policy"] = approval_content_security_policy(
        script_nonce=script_nonce,
    )
    return headers


def _local_proof_copy_script(script_nonce: str) -> str:
    return (
        f'<script nonce="{escape(script_nonce)}">\n'
        "(function () {\n"
        '  document.querySelectorAll(".approval-copy-command").forEach(function (button) {\n'
        '    button.addEventListener("click", function () {\n'
        '      var target = document.getElementById(button.getAttribute("data-copy-target"));\n'
        "      if (!target) { return; }\n"
        '      var text = target.textContent || "";\n'
        "      if (navigator.clipboard && navigator.clipboard.writeText) {\n"
        "        navigator.clipboard.writeText(text).then(function () {\n"
        '          button.textContent = "Copied";\n'
        '          setTimeout(function () { button.textContent = "Copy prompt"; }, 1500);\n'
        "        }).catch(function () {\n"
        "          var range = document.createRange();\n"
        "          range.selectNodeContents(target);\n"
        "          window.getSelection().removeAllRanges();\n"
        "          window.getSelection().addRange(range);\n"
        "        });\n"
        "      } else {\n"
        "        var range = document.createRange();\n"
        "        range.selectNodeContents(target);\n"
        "        window.getSelection().removeAllRanges();\n"
        "        window.getSelection().addRange(range);\n"
        "      }\n"
        "    });\n"
        "  });\n"
        "})();\n"
        "</script>"
    )


def render_local_proof_block(
    body_text: str,
    *,
    script_nonce: str,
    agent_prompt: str | None = None,
) -> str:
    """Return a compact Approval Center block for local proof inspection."""

    prompt = LOCAL_PROOF_AGENT_PROMPT if agent_prompt is None else agent_prompt
    return (
        '<section class="approval-local-proof">'
        f'<h2 class="approval-local-proof-title">{escape(LOCAL_PROOF_BLOCK_TITLE)}</h2>'
        f'<p class="approval-local-proof-body">{escape(body_text)}</p>'
        '<div class="approval-local-proof-command-row">'
        f'<code class="approval-local-proof-command" id="{_LOCAL_PROOF_COMMAND_ELEMENT_ID}">'
        f"{escape(prompt)}</code>"
        f'<button type="button" class="approval-copy-command" '
        f'data-copy-target="{_LOCAL_PROOF_COMMAND_ELEMENT_ID}">'
        f"{escape(LOCAL_PROOF_COPY_COMMAND_LABEL)}</button>"
        "</div>"
        f"{_local_proof_copy_script(script_nonce)}"
        "</section>"
    )


def render_local_proof_prompt_block(body_text: str, prompt: str) -> str:
    """Return a human-facing local proof hint without scripts or command UI."""

    return (
        '<section class="approval-local-proof approval-local-proof-human">'
        f'<h2 class="approval-local-proof-title">{escape(LOCAL_PROOF_BLOCK_TITLE)}</h2>'
        f'<p class="approval-local-proof-body">{escape(body_text)}</p>'
        f'<p class="approval-local-proof-prompt">{escape(prompt)}</p>'
        "</section>"
    )

TERMINAL_ALREADY_DECIDED_APPROVE = "already_decided_approve"
TERMINAL_ALREADY_DECIDED_DENY = "already_decided_deny"
TERMINAL_APPROVAL_EXPIRED = "approval_expired"
TERMINAL_ALREADY_DECIDED = "already_decided"


class _DaemonThreadingHTTPServer(ThreadingHTTPServer):
    daemon_threads = True


class ApprovalServerError(RuntimeError):
    """Raised when the local approval server cannot operate safely."""


class ApprovalServerGone(ApprovalServerError):
    """Raised when a submitted approval URL is no longer actionable."""


@dataclass(frozen=True)
class ApprovalPrompt:
    """Privacy-filtered approval prompt data served by the local UI."""

    request_id: str
    client_id: str
    session_id: str
    downstream_server: str
    tool_name: str
    action_display: str
    action_details: str | None
    resource_display: str | None
    resource_details: str | None
    risk_class: str
    payload_hash: str
    policy_rule_id: str
    reason: str
    created_at: int
    expires_at: int
    csrf_token: str
    action_gate_metadata: dict[str, Any] | None = None
    scope_expansion_allowed: bool = False


@dataclass(frozen=True)
class ApprovalServerDecision:
    """Decision submitted through the local approval server."""

    request_id: str
    decision: str
    approval_scope: str


@dataclass(frozen=True)
class TerminalApprovalSnapshot:
    """Sanitized approval context retained for stale loopback URLs."""

    request_id: str
    state: str
    client_id: str
    session_id_prefix: str
    downstream_server: str
    tool_name: str
    risk_class: str
    reason: str
    action_display: str
    resource_display: str
    policy_rule_id: str
    created_at: int
    expires_at: int

    @classmethod
    def from_prompt(cls, prompt: ApprovalPrompt, *, state: str) -> TerminalApprovalSnapshot:
        return cls(
            request_id=prompt.request_id,
            state=state,
            client_id=prompt.client_id,
            session_id_prefix=prompt.session_id[:8],
            downstream_server=prompt.downstream_server,
            tool_name=prompt.tool_name,
            risk_class=prompt.risk_class,
            reason=prompt.reason,
            action_display=prompt.action_display,
            resource_display=prompt.resource_display or "none",
            policy_rule_id=prompt.policy_rule_id,
            created_at=prompt.created_at,
            expires_at=prompt.expires_at,
        )

    @classmethod
    def from_pending_record(
        cls,
        record: Any,
        *,
        state: str | None = None,
    ) -> TerminalApprovalSnapshot | None:
        """Build a terminal snapshot from one durable evidence record."""

        resolved_state = state or terminal_state_for_record_status(record.status)
        if resolved_state is None:
            return None
        client_id = record.client_id if isinstance(record.client_id, str) and record.client_id else "-"
        session_prefix = record.session_id[:8] if isinstance(record.session_id, str) else "-"
        policy_rule_id = record.policy_rule_id if isinstance(record.policy_rule_id, str) else "-"
        expires_at = record.expires_at if record.expires_at is not None else record.created_at
        return cls(
            request_id=record.request_id,
            state=resolved_state,
            client_id=client_id,
            session_id_prefix=session_prefix,
            downstream_server=record.downstream_server,
            tool_name=record.tool_name,
            risk_class=record.risk_class,
            reason=bounded_reason_for_record(record),
            action_display=bounded_action_display(record),
            resource_display=bounded_resource_display(record),
            policy_rule_id=policy_rule_id,
            created_at=record.created_at,
            expires_at=expires_at,
        )


def approval_prompt_to_dict(prompt: ApprovalPrompt) -> dict[str, Any]:
    """Serialize one approval prompt for the persistent center register API."""

    return {
        "request_id": prompt.request_id,
        "client_id": prompt.client_id,
        "session_id": prompt.session_id,
        "downstream_server": prompt.downstream_server,
        "tool_name": prompt.tool_name,
        "action_display": prompt.action_display,
        "action_details": prompt.action_details,
        "resource_display": prompt.resource_display,
        "resource_details": prompt.resource_details,
        "risk_class": prompt.risk_class,
        "payload_hash": prompt.payload_hash,
        "policy_rule_id": prompt.policy_rule_id,
        "reason": prompt.reason,
        "created_at": prompt.created_at,
        "expires_at": prompt.expires_at,
        "csrf_token": prompt.csrf_token,
        "action_gate_metadata": prompt.action_gate_metadata,
        "scope_expansion_allowed": prompt.scope_expansion_allowed,
    }


def approval_prompt_from_dict(data: Mapping[str, Any]) -> ApprovalPrompt:
    """Deserialize one approval prompt from the persistent center register API."""

    return ApprovalPrompt(
        request_id=str(data["request_id"]),
        client_id=str(data["client_id"]),
        session_id=str(data["session_id"]),
        downstream_server=str(data["downstream_server"]),
        tool_name=str(data["tool_name"]),
        action_display=str(data["action_display"]),
        action_details=data.get("action_details"),
        resource_display=data.get("resource_display"),
        resource_details=data.get("resource_details"),
        risk_class=str(data["risk_class"]),
        payload_hash=str(data["payload_hash"]),
        policy_rule_id=str(data["policy_rule_id"]),
        reason=str(data["reason"]),
        created_at=int(data["created_at"]),
        expires_at=int(data["expires_at"]),
        csrf_token=str(data["csrf_token"]),
        action_gate_metadata=data.get("action_gate_metadata"),
        scope_expansion_allowed=bool(data.get("scope_expansion_allowed", False)),
    )


class ApprovalServer:
    """Authenticated loopback HTTP server for local approval decisions."""

    owns_server_process = True

    def __init__(
        self,
        *,
        host: str = "127.0.0.1",
        port: int = 0,
        session_token: str | None = None,
        internal_register_token: str | None = None,
        evidence_store: Any = None,
    ):
        if host != "127.0.0.1":
            raise ApprovalServerError("approval server must bind to 127.0.0.1")
        self.host = host
        self.port = port
        self.session_token = session_token or secrets.token_urlsafe(32)
        self.internal_register_token = internal_register_token
        self.evidence_store = evidence_store
        self._hmac_key = secrets.token_bytes(32)
        self._cookie_nonce = secrets.token_urlsafe(16)
        self._lock = threading.RLock()
        self._prompts: dict[str, ApprovalPrompt] = {}
        self._decisions: dict[str, ApprovalServerDecision] = {}
        self._decision_events: dict[str, threading.Event] = {}
        self._terminal_requests: dict[str, float] = {}
        self._terminal_snapshots: dict[str, TerminalApprovalSnapshot] = {}
        self._decision_handler: Callable[[ApprovalServerDecision], None] | None = None
        self._httpd: _DaemonThreadingHTTPServer | None = None
        self._thread: threading.Thread | None = None

    def set_decision_handler(
        self,
        handler: Callable[[ApprovalServerDecision], None] | None,
    ) -> None:
        """Register a callback invoked after each approve/deny POST is accepted.

        The handler runs outside the server lock so it can persist evidence
        before the HTTP response returns. That closes the live-console race
        where the client retries immediately after POST while the background
        watcher has not yet written APPROVED to the evidence store.
        """

        self._decision_handler = handler

    @property
    def token_hash(self) -> str:
        """Return a stable hash of the current per-process session token."""

        return "sha256:" + hashlib.sha256(self.session_token.encode("utf-8")).hexdigest()

    @property
    def base_url(self) -> str:
        """Return the loopback base URL."""

        if self._httpd is None:
            raise ApprovalServerError("approval server is not started")
        return f"http://{self.host}:{self.port}"

    @property
    def is_running(self) -> bool:
        """Return whether the HTTP server has been started."""

        return self._httpd is not None

    def start(self) -> None:
        """Start the loopback approval server in a background thread."""

        if self._httpd is not None:
            return

        owner = self

        class Handler(_ApprovalRequestHandler):
            server_owner = owner

        self._httpd = _DaemonThreadingHTTPServer((self.host, self.port), Handler)
        self.host, self.port = self._httpd.server_address[:2]
        self._thread = threading.Thread(
            target=self._httpd.serve_forever,
            name="agentveil-mcp-proxy-approval-server",
            daemon=True,
        )
        self._thread.start()

    def stop(self, *, timeout: float = 2.0) -> None:
        """Stop the approval server."""

        httpd = self._httpd
        if httpd is None:
            return
        httpd.shutdown()
        httpd.server_close()
        if self._thread is not None:
            self._thread.join(timeout=timeout)
        self._httpd = None
        self._thread = None

    def approval_center_url(self) -> str:
        """Return the loopback approval list URL for the current server session."""

        return f"{self.base_url}/approval/{self.session_token}"

    def approval_url(self, request_id: str) -> str:
        """Return the authenticated approval URL for a pending request."""

        return f"{self.base_url}/approval/{self.session_token}/pending/{request_id}"

    def register(self, prompt: ApprovalPrompt) -> str:
        """Register a prompt after durable evidence persistence succeeds."""

        with self._lock:
            self._prune_terminal_requests_locked()
            self._prompts[prompt.request_id] = prompt
            self._decision_events[prompt.request_id] = threading.Event()
            self._terminal_requests.pop(prompt.request_id, None)
            self._terminal_snapshots.pop(prompt.request_id, None)
        return self.approval_url(prompt.request_id)

    def unregister(self, request_id: str, *, terminal_state: str = TERMINAL_ALREADY_DECIDED) -> None:
        """Mark an approval URL as no longer actionable."""

        with self._lock:
            self._prune_terminal_requests_locked()
            prompt = self._prompts.pop(request_id, None)
            decision = self._decisions.pop(request_id, None)
            self._decision_events.pop(request_id, None)
            state = terminal_state
            if prompt is not None and state == TERMINAL_ALREADY_DECIDED and decision is not None:
                if decision.decision == "approve":
                    state = TERMINAL_ALREADY_DECIDED_APPROVE
                elif decision.decision == "deny":
                    state = TERMINAL_ALREADY_DECIDED_DENY
            if prompt is not None:
                self._terminal_snapshots[request_id] = TerminalApprovalSnapshot.from_prompt(
                    prompt,
                    state=state,
                )
            self._terminal_requests[request_id] = self._terminal_retain_until(prompt)

    def list_decided_request_ids(self) -> tuple[str, ...]:
        """Return request IDs with a local approve/deny decision not yet unregistered."""

        with self._lock:
            return tuple(self._decisions.keys())

    def get_decision(self, request_id: str) -> ApprovalServerDecision | None:
        """Return a recorded local decision, if present."""

        with self._lock:
            return self._decisions.get(request_id)

    def wait_for_decision(self, request_id: str, *, timeout: float) -> ApprovalServerDecision | None:
        """Wait for an approve/deny POST for one request."""

        with self._lock:
            event = self._decision_events.get(request_id)
        if event is None:
            return None
        if not event.wait(timeout=timeout):
            return None
        with self._lock:
            return self._decisions.get(request_id)

    def pending_row_dict(self, prompt: ApprovalPrompt) -> dict[str, Any]:
        """Sanitized pending row shared by the dashboard HTML and JSON API."""

        return pending_approval_dict(
            request_id=prompt.request_id,
            client_id=prompt.client_id,
            session_id=prompt.session_id,
            downstream_server=prompt.downstream_server,
            tool_name=prompt.tool_name,
            action_display=prompt.action_display,
            resource_display=prompt.resource_display,
            risk_class=prompt.risk_class,
            reason=prompt.reason,
            payload_hash=prompt.payload_hash,
            policy_rule_id=prompt.policy_rule_id,
            created_at=prompt.created_at,
            expires_at=prompt.expires_at,
            action_gate_metadata=prompt.action_gate_metadata,
        )

    def pending_approvals_api_payload(self) -> dict[str, Any]:
        """Return sanitized pending approvals for the loopback JSON API."""

        return {
            "ok": True,
            "approvals": [
                self.pending_row_dict(prompt)
                for prompt in self.pending_prompts()
            ],
        }

    def pending_prompts(self) -> list[ApprovalPrompt]:
        """Return currently pending prompts for the token-authenticated list page."""

        with self._lock:
            self._prune_terminal_requests_locked()
            self._prune_expired_pending_locked()
            decided = set(self._decisions)
            terminal = set(self._terminal_requests)
            return [
                prompt
                for request_id, prompt in sorted(self._prompts.items())
                if request_id not in decided and request_id not in terminal
            ]

    def prompt_for(self, request_id: str) -> ApprovalPrompt | None:
        """Return one pending prompt."""

        with self._lock:
            self._prune_terminal_requests_locked()
            if request_id in self._decisions or request_id in self._terminal_requests:
                return None
            return self._prompts.get(request_id)

    def is_terminal(self, request_id: str) -> bool:
        """Return whether a prompt was already decided or expired."""

        with self._lock:
            self._prune_terminal_requests_locked()
            return request_id in self._terminal_requests or request_id in self._decisions

    def terminal_snapshot_for(self, request_id: str) -> TerminalApprovalSnapshot | None:
        """Return sanitized stale-state context while the terminal retention window is active."""

        with self._lock:
            self._prune_terminal_requests_locked()
            return self._terminal_snapshots.get(request_id)

    def stale_terminal_snapshot_for(self, request_id: str) -> TerminalApprovalSnapshot | None:
        """Return retained or inferred terminal context for a non-actionable request."""

        with self._lock:
            self._prune_terminal_requests_locked()
            snapshot = self._terminal_snapshots.get(request_id)
            if snapshot is not None:
                return snapshot
            decision = self._decisions.get(request_id)
            prompt = self._prompts.get(request_id)
            if decision is not None and prompt is not None:
                state = TERMINAL_ALREADY_DECIDED
                if decision.decision == "approve":
                    state = TERMINAL_ALREADY_DECIDED_APPROVE
                elif decision.decision == "deny":
                    state = TERMINAL_ALREADY_DECIDED_DENY
                return TerminalApprovalSnapshot.from_prompt(prompt, state=state)
        return self._terminal_snapshot_from_evidence(request_id)

    def _terminal_snapshot_from_evidence(self, request_id: str) -> TerminalApprovalSnapshot | None:
        """Load terminal context from durable evidence when in-memory state is gone."""

        store = self.evidence_store
        if store is None:
            return None
        record = store.get_pending(request_id)
        if record is None:
            return None
        return TerminalApprovalSnapshot.from_pending_record(record)

    def submit_decision(self, request_id: str, decision: str, approval_scope: str) -> None:
        """Record a local approve/deny POST."""

        if decision not in {"approve", "deny"}:
            raise ApprovalServerError("approval decision must be approve or deny")
        if approval_scope not in {"exact", "similar_5m"}:
            raise ApprovalServerError("approval scope is unsupported")
        with self._lock:
            self._prune_terminal_requests_locked()
            if request_id in self._terminal_requests or request_id in self._decisions:
                raise ApprovalServerGone("approval already decided")
            prompt = self._prompts.get(request_id)
            if prompt is None:
                raise ApprovalServerError("pending approval not found")
            if approval_scope == "similar_5m" and not prompt.scope_expansion_allowed:
                raise ApprovalServerError("approval scope is not available for this request")
            self._decisions[request_id] = ApprovalServerDecision(
                request_id=request_id,
                decision=decision,
                approval_scope=approval_scope,
            )
            self._decision_events.setdefault(request_id, threading.Event()).set()
            snapshot = self._decisions[request_id]
            handler = self._decision_handler
        if handler is not None:
            handler(snapshot)

    def _cookie_value(self) -> str:
        message = f"{self.session_token}:{self._cookie_nonce}".encode("utf-8")
        return hmac.new(self._hmac_key, message, hashlib.sha256).hexdigest()

    def _valid_cookie(self, raw_cookie: str | None) -> bool:
        if not raw_cookie:
            return False
        cookie = SimpleCookie()
        try:
            cookie.load(raw_cookie)
        except Exception:
            return False
        morsel = cookie.get(COOKIE_NAME)
        if morsel is None:
            return False
        return hmac.compare_digest(morsel.value, self._cookie_value())

    def _valid_internal_register_token(self, raw_token: str | None) -> bool:
        expected = self.internal_register_token
        if not isinstance(expected, str) or not expected:
            return False
        if not isinstance(raw_token, str) or not raw_token:
            return False
        return hmac.compare_digest(raw_token, expected)

    def _terminal_retain_until(self, prompt: ApprovalPrompt | None) -> float:
        now = time.time()
        retention = DEFAULT_TERMINAL_REQUEST_RETENTION_SECONDS
        if prompt is not None:
            retention = max(
                float(prompt.expires_at - prompt.created_at) * 2.0,
                MIN_TERMINAL_REQUEST_RETENTION_SECONDS,
            )
        return now + retention

    def _prune_terminal_requests_locked(self) -> None:
        now = time.time()
        expired = [
            request_id
            for request_id, retain_until in self._terminal_requests.items()
            if retain_until <= now
        ]
        for request_id in expired:
            self._terminal_requests.pop(request_id, None)
            self._terminal_snapshots.pop(request_id, None)

    def _prune_expired_pending_locked(self) -> None:
        """Hide expired pending prompts from the default list without deleting evidence."""

        now = int(time.time())
        expired_ids = [
            request_id
            for request_id, prompt in self._prompts.items()
            if request_id not in self._decisions
            and request_id not in self._terminal_requests
            and now > prompt.expires_at
        ]
        for request_id in expired_ids:
            prompt = self._prompts.pop(request_id, None)
            self._decision_events.pop(request_id, None)
            if prompt is None:
                continue
            self._terminal_snapshots[request_id] = TerminalApprovalSnapshot.from_prompt(
                prompt,
                state=TERMINAL_APPROVAL_EXPIRED,
            )
            self._terminal_requests[request_id] = self._terminal_retain_until(prompt)


class _ApprovalRequestHandler(BaseHTTPRequestHandler):
    server_owner: ApprovalServer

    def setup(self) -> None:
        super().setup()
        self.request.settimeout(REQUEST_SOCKET_TIMEOUT_SECONDS)

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

    def do_GET(self) -> None:
        token, route, request_id = self._parse_path()
        if not self._token_ok(token):
            if route == "api_list":
                self._send_text(HTTPStatus.FORBIDDEN, "forbidden")
            else:
                self._send_html(HTTPStatus.FORBIDDEN, self._render_forbidden_session())
            return
        if route == "api_list":
            self._send_json(HTTPStatus.OK, self.server_owner.pending_approvals_api_payload())
            return
        if route == "list":
            prompts = self.server_owner.pending_prompts()
            if len(prompts) == 1:
                token = self.server_owner.session_token
                location = f"/approval/{token}/pending/{prompts[0].request_id}"
                self._send_redirect(
                    location,
                    extra_headers={"Set-Cookie": self._session_cookie_header()},
                )
                return
            extra_headers = None
            if prompts:
                extra_headers = {"Set-Cookie": self._session_cookie_header()}
            self._send_html(HTTPStatus.OK, self._render_list(), extra_headers=extra_headers)
            return
        if route != "pending" or request_id is None:
            self._send_text(HTTPStatus.NOT_FOUND, "not found")
            return
        self._respond_pending_get(request_id)

    def do_POST(self) -> None:
        token, route, request_id = self._parse_path()
        if route == "internal_register":
            self._handle_internal_register()
            return
        if route == "legacy_internal_register":
            self._send_text(HTTPStatus.FORBIDDEN, "forbidden")
            return
        if route != "pending" or request_id is None or not self._token_ok(token):
            if route == "api_list":
                self._send_text(HTTPStatus.FORBIDDEN, "forbidden")
            else:
                self._send_html(HTTPStatus.FORBIDDEN, self._render_forbidden_session())
            return
        prompt = self.server_owner.prompt_for(request_id)
        if prompt is None:
            self._respond_stale_pending(request_id)
            return
        if int(time.time()) > prompt.expires_at:
            snapshot = TerminalApprovalSnapshot.from_prompt(
                prompt,
                state=TERMINAL_APPROVAL_EXPIRED,
            )
            self._send_html(HTTPStatus.GONE, self._render_terminal(snapshot))
            return
        if not self.server_owner._valid_cookie(self.headers.get("Cookie")):
            self._send_text(HTTPStatus.FORBIDDEN, "forbidden")
            return
        raw_length = self.headers.get("Content-Length", "0") or "0"
        try:
            length = int(raw_length)
        except ValueError:
            self._send_text(HTTPStatus.BAD_REQUEST, "invalid content length")
            return
        if length < 0 or length > MAX_POST_BODY_BYTES:
            self._send_text(HTTPStatus.BAD_REQUEST, "invalid content length")
            return
        body = self.rfile.read(length).decode("utf-8", errors="replace")
        form = parse_qs(body, keep_blank_values=True)
        csrf = (form.get("csrf_token") or [""])[0]
        if not hmac.compare_digest(csrf, prompt.csrf_token):
            self._send_text(HTTPStatus.FORBIDDEN, "forbidden")
            return
        decision = (form.get("decision") or [""])[0]
        scope = (form.get("approval_scope") or ["exact"])[0]
        try:
            self.server_owner.submit_decision(request_id, decision, scope)
        except ApprovalServerGone:
            self._respond_stale_pending(request_id)
            return
        except ApprovalServerError:
            self._send_html(HTTPStatus.FORBIDDEN, self._render_forbidden_session())
            return
        recorded_body = (
            APPROVAL_POST_APPROVE_HUMAN_BODY
            if decision == "approve"
            else APPROVAL_DECISION_DENIED_BODY
        )
        if decision == "approve":
            page_body = (
                f"<p>{escape(recorded_body)}</p>"
                f"{render_local_proof_prompt_block(APPROVAL_POST_APPROVE_PROOF_BODY, APPROVAL_POST_APPROVE_PROOF_PROMPT)}"
            )
            script_nonce = None
        else:
            proof_body = LOCAL_PROOF_POST_DENY_BODY
            script_nonce = generate_approval_script_nonce()
            page_body = (
                f"<p>{escape(APPROVAL_DECISION_DENIED_BODY)}</p>"
                f"{render_local_proof_block(proof_body, script_nonce=script_nonce)}"
            )
        self._send_html(
            HTTPStatus.OK,
            self._page(
                "Approval recorded",
                page_body,
                page_kind="detail",
                include_card_styles=False,
                include_local_proof_styles=True,
            ),
            script_nonce=script_nonce,
        )

    def _respond_pending_get(self, request_id: str) -> None:
        prompt = self.server_owner.prompt_for(request_id)
        if prompt is not None and int(time.time()) > prompt.expires_at:
            snapshot = TerminalApprovalSnapshot.from_prompt(
                prompt,
                state=TERMINAL_APPROVAL_EXPIRED,
            )
            self._send_html(HTTPStatus.GONE, self._render_terminal(snapshot))
            return
        if prompt is not None:
            self._send_html(
                HTTPStatus.OK,
                self._render_prompt(prompt),
                extra_headers={"Set-Cookie": self._session_cookie_header()},
            )
            return
        self._respond_stale_pending(request_id)

    def _respond_stale_pending(self, request_id: str) -> None:
        snapshot = self.server_owner.stale_terminal_snapshot_for(request_id)
        if snapshot is not None:
            self._send_html(HTTPStatus.GONE, self._render_terminal(snapshot))
            return
        if self.server_owner.is_terminal(request_id):
            self._send_html(
                HTTPStatus.GONE,
                self._render_terminal(
                    TerminalApprovalSnapshot(
                        request_id=request_id,
                        state=TERMINAL_ALREADY_DECIDED,
                        client_id="-",
                        session_id_prefix="-",
                        downstream_server="-",
                        tool_name="-",
                        risk_class="-",
                        reason="-",
                        action_display="-",
                        resource_display="none",
                        policy_rule_id="-",
                        created_at=0,
                        expires_at=0,
                    ),
                ),
            )
            return
        self._send_html(HTTPStatus.NOT_FOUND, self._render_request_not_found(request_id))

    def _parse_path(self) -> tuple[str | None, str | None, str | None]:
        parts = [part for part in self.path.split("?")[0].split("/") if part]
        if len(parts) == 2 and parts[0] == "internal" and parts[1] == "register":
            return None, "internal_register", None
        if len(parts) >= 2 and parts[0] == "approval":
            token = parts[1]
            if len(parts) == 2:
                return token, "list", None
            if len(parts) == 4 and parts[2] == "api" and parts[3] == "approvals":
                return token, "api_list", None
            if len(parts) == 4 and parts[2] == "internal" and parts[3] == "register":
                return token, "legacy_internal_register", None
            if len(parts) == 4 and parts[2] == "pending":
                return token, "pending", parts[3]
        return None, None, None

    def _handle_internal_register(self) -> None:
        if not self.server_owner._valid_internal_register_token(
            self.headers.get(INTERNAL_REGISTER_TOKEN_HEADER),
        ):
            self._send_text(HTTPStatus.FORBIDDEN, "forbidden")
            return
        raw_length = self.headers.get("Content-Length", "0") or "0"
        try:
            length = int(raw_length)
        except ValueError:
            self._send_text(HTTPStatus.BAD_REQUEST, "invalid content length")
            return
        if length < 0 or length > MAX_POST_BODY_BYTES:
            self._send_text(HTTPStatus.BAD_REQUEST, "invalid content length")
            return
        body = self.rfile.read(length).decode("utf-8", errors="replace")
        try:
            payload = json.loads(body)
        except json.JSONDecodeError:
            self._send_text(HTTPStatus.BAD_REQUEST, "invalid json")
            return
        if not isinstance(payload, dict):
            self._send_text(HTTPStatus.BAD_REQUEST, "invalid json")
            return
        try:
            prompt = approval_prompt_from_dict(payload)
        except (KeyError, TypeError, ValueError):
            self._send_text(HTTPStatus.BAD_REQUEST, "invalid prompt")
            return
        approval_url = self.server_owner.register(prompt)
        self._send_json(
            HTTPStatus.OK,
            {"ok": True, "approval_url": approval_url},
        )

    def _token_ok(self, token: str | None) -> bool:
        return bool(token) and hmac.compare_digest(token or "", self.server_owner.session_token)

    def _session_cookie_header(self) -> str:
        return (
            f"{COOKIE_NAME}={self.server_owner._cookie_value()}; "
            f"Path=/approval/{self.server_owner.session_token}; HttpOnly; SameSite=Strict"
        )

    def _render_forbidden_session(self) -> str:
        return self._page(
            "Forbidden",
            "<p>Invalid approval session URL.</p>",
        )

    def _render_request_not_found(self, request_id: str) -> str:
        body = (
            "<p>This approval request is no longer pending.</p>"
            f"<p class=\"approval-request-id\">request {escape(request_id)}</p>"
            "<p>Open your Approval Center pending list from the original notification link.</p>"
        )
        return self._page("Request no longer pending", body)

    def _terminal_titles(self, state: str) -> tuple[str, str]:
        if state == TERMINAL_APPROVAL_EXPIRED:
            return "Approval expired", TERMINAL_EXPIRED_BODY
        if state == TERMINAL_ALREADY_DECIDED_APPROVE:
            return "Approved", TERMINAL_ALREADY_APPROVED_BODY
        if state == TERMINAL_ALREADY_DECIDED_DENY:
            return "Denied", TERMINAL_ALREADY_DENIED_BODY
        return "Already decided", TERMINAL_NO_LONGER_ACTIONABLE_BODY

    def _render_terminal_technical_details(self, snapshot: TerminalApprovalSnapshot) -> str:
        items = (
            f"<dt>Request</dt><dd>{escape(snapshot.request_id)}</dd>"
            f"<dt>Client</dt><dd>{escape(snapshot.client_id)}</dd>"
            f"<dt>Session prefix</dt><dd>{escape(snapshot.session_id_prefix)}</dd>"
            f"<dt>Downstream</dt><dd>{escape(snapshot.downstream_server)}</dd>"
            f"<dt>Tool</dt><dd>{escape(snapshot.tool_name)}</dd>"
            f"<dt>Action</dt><dd>{escape(snapshot.action_display)}</dd>"
            f"<dt>Resource</dt><dd>{escape(snapshot.resource_display)}</dd>"
            f"<dt>Risk</dt><dd>{escape(snapshot.risk_class)}</dd>"
            f"<dt>Reason</dt><dd>{escape(snapshot.reason)}</dd>"
            f"<dt>Policy rule</dt><dd>{escape(snapshot.policy_rule_id)}</dd>"
            f"<dt>Created</dt><dd>{snapshot.created_at}</dd>"
            f"<dt>Expires</dt><dd>{snapshot.expires_at}</dd>"
        )
        return (
            '<details class="approval-proof-details">'
            "<summary>Technical details</summary>"
            f'<dl class="approval-detail">{items}</dl>'
            "</details>"
        )

    def _render_terminal(self, snapshot: TerminalApprovalSnapshot) -> str:
        title, lead = self._terminal_titles(snapshot.state)
        body_parts = [f"<p>{escape(lead)}</p>"]
        if snapshot.state == TERMINAL_ALREADY_DECIDED_APPROVE:
            body_parts.append(f"<p>{escape(TERMINAL_ALREADY_APPROVED_NEXT)}</p>")
        body_parts.append(
            render_local_proof_prompt_block(
                APPROVAL_POST_APPROVE_PROOF_BODY,
                APPROVAL_POST_APPROVE_PROOF_PROMPT,
            )
        )
        body_parts.append(self._render_terminal_technical_details(snapshot))
        return self._page(
            title,
            "".join(body_parts),
            page_kind="detail",
            include_card_styles=False,
            include_local_proof_styles=True,
        )

    def _render_list(self) -> str:
        prompts = self.server_owner.pending_prompts()
        token = self.server_owner.session_token
        if not prompts:
            body = '<p class="approval-empty">No pending approvals</p>'
        else:
            cards = "".join(
                self._render_dashboard_card(
                    self.server_owner.pending_row_dict(prompt),
                    detail_href=f"/approval/{token}/pending/{prompt.request_id}",
                    csrf_token=prompt.csrf_token,
                )
                for prompt in prompts
            )
            body = f'<section class="approval-cards">{cards}</section>'
        return self._page(
            "Pending approvals",
            body,
            page_kind="dashboard",
            include_card_styles=bool(prompts),
        )

    def _render_dashboard_card(
        self,
        row: dict[str, Any],
        *,
        detail_href: str,
        csrf_token: str,
    ) -> str:
        risk = escape(row["risk_class"])
        risk_label = escape(risk_class_plain_label(str(row["risk_class"])))
        summary = escape(
            human_approval_summary(
                tool_name=str(row["tool_name"]),
                resource_display=str(row.get("resource")),
            )
        )
        reason_label = escape(human_approval_reason_label(str(row.get("reason", ""))))
        decision_form = (
            f'<form method="post" action="{escape(detail_href)}">'
            f'<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">'
            '<input type="hidden" name="approval_scope" value="exact">'
            '<button type="submit" name="decision" value="approve">Approve</button>'
            '<button type="submit" name="decision" value="deny">Deny</button>'
            "</form>"
        )
        return (
            '<article class="approval-card">'
            '<header class="approval-card-header">'
            f'<span class="approval-tool">{summary}</span>'
            f'<span class="approval-risk approval-risk-{risk}">{risk_label}</span>'
            "</header>"
            f'<p class="approval-reason">{reason_label}</p>'
            '<dl class="approval-meta">'
            f'<div><dt>Tool</dt><dd class="approval-wrap">{escape(row["tool_name"])}</dd></div>'
            f'<div><dt>Target</dt><dd class="approval-wrap">{escape(row["resource"])}</dd></div>'
            f'<div><dt>Client</dt><dd class="approval-wrap">{escape(row["client_id"])}</dd></div>'
            "</dl>"
            f'<p class="approval-card-actions">'
            f'<a class="approval-button" href="{escape(detail_href)}">Review &amp; decide</a>'
            "</p>"
            f"{decision_form}"
            "</article>"
        )

    def _render_prompt(self, prompt: ApprovalPrompt) -> str:
        token = self.server_owner.session_token
        summary = human_approval_summary(
            tool_name=prompt.tool_name,
            resource_display=prompt.resource_display,
        )
        risk_label = risk_class_plain_label(prompt.risk_class)
        reason_label = human_approval_reason_label(prompt.reason)
        title = f"Review: {prompt.tool_name}"
        back_link = (
            f'<p class="approval-back"><a href="/approval/{escape(token)}">'
            "&larr; Back to pending list</a></p>"
        )
        detail = ""
        if prompt.action_details or prompt.resource_details:
            detail_items = []
            if prompt.action_details:
                detail_items.append(f"<dt>Action detail</dt><dd>{escape(prompt.action_details)}</dd>")
            if prompt.resource_details:
                detail_items.append(f"<dt>Resource detail</dt><dd>{escape(prompt.resource_details)}</dd>")
            detail = (
                "<details><summary>Show local details</summary><dl>"
                + "".join(detail_items)
                + "</dl></details>"
            )
        similar = ""
        if prompt.scope_expansion_allowed:
            similar = (
                "<form method=\"post\">"
                f"<input type=\"hidden\" name=\"csrf_token\" value=\"{escape(prompt.csrf_token)}\">"
                "<input type=\"hidden\" name=\"approval_scope\" value=\"similar_5m\">"
                "<button type=\"submit\" name=\"decision\" value=\"approve\">"
                "Approve similar for 5 minutes</button>"
                "</form>"
            )
        session_prefix = prompt.session_id[:8]
        proof_rows = approval_proof_detail_rows(
            tool_name=prompt.tool_name,
            resource_display=prompt.resource_display,
            risk_class=prompt.risk_class,
            reason=prompt.reason,
            payload_hash=prompt.payload_hash,
            policy_rule_id=prompt.policy_rule_id,
            request_id=prompt.request_id,
            created_at=prompt.created_at,
            expires_at=prompt.expires_at,
            action_gate_metadata=prompt.action_gate_metadata,
        )
        proof_items = "".join(
            f"<dt>{escape(label)}</dt><dd>{escape(value)}</dd>"
            for label, value in proof_rows
        )
        raw_rows = approval_raw_evidence_rows(
            client_id=prompt.client_id,
            session_id_prefix=session_prefix,
            action_display=prompt.action_display,
            action_gate_metadata=prompt.action_gate_metadata,
        )
        raw_items = "".join(
            f"<dt>{escape(label)}</dt><dd>{escape(value)}</dd>"
            for label, value in raw_rows
        )
        body = f"""
{back_link}
<p class="approval-summary">{escape(summary)}</p>
<p class="approval-reason">{escape(reason_label)}</p>
<dl class="approval-detail">
<dt>Tool</dt><dd>{escape(prompt.tool_name)}</dd>
<dt>Target</dt><dd>{escape(prompt.resource_display or "none")}</dd>
<dt>Risk</dt><dd>{escape(risk_label)}</dd>
</dl>
<form method=\"post\">
<input type=\"hidden\" name=\"csrf_token\" value=\"{escape(prompt.csrf_token)}\">
<input type=\"hidden\" name=\"approval_scope\" value=\"exact\">
<button type=\"submit\" name=\"decision\" value=\"approve\">Approve</button>
<button type=\"submit\" name=\"decision\" value=\"deny\">Deny</button>
</form>
<p class="approval-decision-note">{escape(LOCAL_PROOF_PENDING_QUIET_LINE)}</p>
{similar}
<details class="approval-proof-details">
<summary>Proof details</summary>
<dl class="approval-detail">
{proof_items}
</dl>
<details class="approval-raw-evidence">
<summary>Raw evidence</summary>
<dl class="approval-detail">
{raw_items}
</dl>
{detail}
</details>
</details>
"""
        return self._page(title, body, page_kind="detail")

    def _security_notice_html(self) -> str:
        return (
            '<p class="approval-security-notice">'
            f"{escape(APPROVAL_LOCAL_URL_WARNING)}"
            "</p>"
        )

    def _approval_page_styles(
        self,
        *,
        include_card_styles: bool = True,
        include_local_proof_styles: bool = False,
    ) -> str:
        card_styles = """
.approval-cards { display: flex; flex-direction: column; gap: 8px; }
.approval-card {
  border: 1px solid var(--border);
  border-radius: 8px;
  padding: 12px;
  background: var(--card-bg);
}
.approval-card-header {
  display: flex;
  justify-content: space-between;
  align-items: flex-start;
  gap: 8px;
  margin-bottom: 8px;
}
.approval-tool { font-weight: 600; word-break: break-word; color: var(--text); }
.approval-risk {
  flex-shrink: 0;
  font-size: 11px;
  font-weight: 600;
  text-transform: uppercase;
  letter-spacing: 0.02em;
  padding: 2px 6px;
  border-radius: 4px;
  background: var(--badge-bg);
  color: var(--badge-text);
}
.approval-reason {
  margin: 0 0 8px;
  color: var(--muted);
  word-break: break-word;
  overflow-wrap: anywhere;
}
.approval-meta {
  margin: 0 0 8px;
  display: grid;
  gap: 4px 12px;
  grid-template-columns: auto 1fr;
}
.approval-meta div { display: contents; }
.approval-meta dt {
  margin: 0;
  color: var(--label);
  font-size: 11px;
  text-transform: uppercase;
}
.approval-meta dd { margin: 0; color: var(--text-secondary); }
.approval-wrap {
  word-break: break-word;
  overflow-wrap: anywhere;
}
.approval-request-id {
  margin: 0 0 8px;
  font-size: 11px;
  color: var(--label);
  font-family: ui-monospace, SFMono-Regular, Menlo, monospace;
  overflow-wrap: anywhere;
}
.approval-card-actions { margin: 0; }
.approval-button {
  display: inline-block;
  padding: 6px 10px;
  border-radius: 6px;
  border: 1px solid var(--button-border);
  background: var(--button-bg);
  color: var(--button-text);
  text-decoration: none;
  font-size: 12px;
  font-weight: 500;
}
.approval-button:hover { background: var(--button-bg-hover); }
""" if include_card_styles else ""
        return f"""
<style>
:root {{
  --bg: #12141a;
  --text: #e8eaed;
  --text-secondary: #d1d5db;
  --muted: #b8bcc4;
  --label: #6b7280;
  --border: #2d333b;
  --card-bg: #1a1d24;
  --badge-bg: #2d333b;
  --badge-text: #c9d1d9;
  --button-bg: #252930;
  --button-bg-hover: #2f3540;
  --button-border: #3d4450;
  --button-text: #e8eaed;
  --link: #8ab4f8;
  --notice-bg: #1a1d24;
  --notice-border: #2d333b;
  --notice-text: #9aa0a6;
}}
@media (prefers-color-scheme: light) {{
  :root {{
    --bg: #f6f7f9;
    --text: #1f2328;
    --text-secondary: #3d444d;
    --muted: #57606a;
    --label: #656d76;
    --border: #d0d7de;
    --card-bg: #ffffff;
    --badge-bg: #eaeef2;
    --badge-text: #24292f;
    --button-bg: #f6f8fa;
    --button-bg-hover: #eaeef2;
    --button-border: #d0d7de;
    --button-text: #24292f;
    --link: #0969da;
    --notice-bg: #f6f8fa;
    --notice-border: #d0d7de;
    --notice-text: #57606a;
  }}
}}
body {{
  margin: 0;
  padding: 16px;
  font: 13px/1.45 system-ui, -apple-system, Segoe UI, sans-serif;
  color: var(--text);
  background: var(--bg);
}}
h1 {{ font-size: 16px; font-weight: 600; margin: 0 0 12px; color: var(--text); }}
.approval-security-notice {{
  margin: 0 0 12px;
  padding: 8px 10px;
  border-radius: 6px;
  border: 1px solid var(--notice-border);
  background: var(--notice-bg);
  color: var(--notice-text);
  font-size: 12px;
  line-height: 1.4;
}}
.approval-empty {{ margin: 0; color: var(--muted); }}
.approval-decision-note {{
  margin: 0 0 12px;
  color: var(--muted);
  font-size: 12px;
  line-height: 1.4;
}}
{self._local_proof_block_styles() if include_local_proof_styles else ""}{card_styles}.approval-back {{ margin: 0 0 12px; font-size: 12px; }}
.approval-back a {{ color: var(--link); text-decoration: none; }}
dl {{ margin: 0 0 12px; }}
dt {{ color: var(--label); font-size: 11px; text-transform: uppercase; }}
dd {{ margin: 0 0 8px; color: var(--text-secondary); word-break: break-word; overflow-wrap: anywhere; }}
button {{
  margin-right: 8px;
  padding: 6px 10px;
  border-radius: 6px;
  border: 1px solid var(--button-border);
  background: var(--button-bg);
  color: var(--button-text);
  font-size: 12px;
  cursor: pointer;
}}
details {{ margin: 8px 0 12px; }}
</style>
"""

    @staticmethod
    def _local_proof_block_styles() -> str:
        return """
.approval-local-proof {
  margin: 12px 0;
  padding: 10px 12px;
  border: 1px solid var(--border);
  border-radius: 8px;
  background: var(--card-bg);
}
.approval-local-proof-title {
  margin: 0 0 6px;
  font-size: 12px;
  font-weight: 600;
  letter-spacing: 0.02em;
  text-transform: uppercase;
  color: var(--text);
}
.approval-local-proof-body {
  margin: 0 0 8px;
  color: var(--muted);
  font-size: 12px;
  line-height: 1.4;
}
.approval-local-proof-command {
  display: block;
  flex: 1 1 auto;
  padding: 8px 10px;
  border-radius: 6px;
  border: 1px solid var(--border);
  background: var(--bg);
  color: var(--text);
  font-family: ui-monospace, SFMono-Regular, Menlo, monospace;
  font-size: 12px;
  white-space: pre-wrap;
  word-break: break-word;
  /* claim-check: allow CSS user-select value, not a product coverage claim. */
  user-select: all;
}
.approval-local-proof-command-row {
  display: flex;
  flex-wrap: wrap;
  gap: 8px;
  align-items: stretch;
}
.approval-copy-command {
  flex: 0 0 auto;
  align-self: stretch;
}
.approval-local-proof-human {
  background: var(--bg);
}
.approval-local-proof-prompt {
  margin: 0;
  padding: 8px 10px;
  border-radius: 6px;
  border: 1px solid var(--border);
  background: var(--card-bg);
  color: var(--text);
  font-size: 13px;
  line-height: 1.4;
}
"""

    def _page(
        self,
        title: str,
        body: str,
        *,
        page_kind: str = "plain",
        include_card_styles: bool = True,
        include_local_proof_styles: bool = False,
    ) -> str:
        styles = (
            self._approval_page_styles(
                include_card_styles=include_card_styles,
                include_local_proof_styles=include_local_proof_styles,
            )
            if page_kind in {"dashboard", "detail"}
            else ""
        )
        notice = (
            self._security_notice_html()
            if page_kind in {"dashboard", "detail"}
            else ""
        )
        return (
            "<!doctype html><html><head><meta charset=\"utf-8\">"
            f"<title>{escape(title)}</title>{styles}</head>"
            f"<body>{notice}<h1>{escape(title)}</h1>{body}</body></html>"
        )

    def _send_redirect(
        self,
        location: str,
        *,
        extra_headers: dict[str, str] | None = None,
    ) -> None:
        self.send_response(int(HTTPStatus.FOUND))
        for key, value in approval_security_headers().items():
            self.send_header(key, value)
        self.send_header("Location", location)
        self.send_header("Content-Length", "0")
        self.send_header("Connection", "close")
        if extra_headers:
            for key, value in extra_headers.items():
                self.send_header(key, value)
        self.end_headers()
        self.close_connection = True

    def _send_html(
        self,
        status: HTTPStatus,
        body: str,
        *,
        extra_headers: dict[str, str] | None = None,
        script_nonce: str | None = None,
    ) -> None:
        self._send_bytes(
            status,
            body.encode("utf-8"),
            "text/html; charset=utf-8",
            extra_headers,
            script_nonce=script_nonce,
        )

    def _send_text(self, status: HTTPStatus, body: str) -> None:
        self._send_bytes(status, body.encode("utf-8"), "text/plain; charset=utf-8", None)

    def _send_json(self, status: HTTPStatus, payload: dict[str, Any]) -> None:
        self._send_bytes(
            status,
            json.dumps(payload, separators=(",", ":")).encode("utf-8"),
            "application/json; charset=utf-8",
            None,
        )

    def _send_bytes(
        self,
        status: HTTPStatus,
        body: bytes,
        content_type: str,
        extra_headers: dict[str, str] | None,
        *,
        script_nonce: str | None = None,
    ) -> None:
        self.send_response(int(status))
        for key, value in approval_security_headers(script_nonce=script_nonce).items():
            self.send_header(key, value)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Connection", "close")
        if extra_headers:
            for key, value in extra_headers.items():
                self.send_header(key, value)
        self.end_headers()
        self.wfile.write(body)
        self.wfile.flush()
        self.close_connection = True


_MANAGED_CENTER_START_TIMEOUT_SECONDS = 60.0
_MANAGED_CENTER_POLL_INTERVAL_SECONDS = 0.2
_MANAGED_CENTER_STOP_GRACE_SECONDS = 2.0


@dataclass(frozen=True)
class ManagedApprovalCenterStatus:
    """Bounded managed Approval Center status for launcher/setup reporting."""

    state: str  # running | stale | down
    pid: int | None
    port: int | None


@dataclass(frozen=True)
class ManagedApprovalCenterEnsureResult:
    status: ManagedApprovalCenterStatus
    started: bool
    reused: bool
    restarted: bool
    reason: str


def _managed_center_proxy_dir(home: Path) -> Path:
    return Path(home) / "mcp-proxy"


def inspect_managed_approval_center(home: Path) -> ManagedApprovalCenterStatus:
    from agentveil_mcp_proxy.approval.persistent import (
        load_manifest,
        manifest_is_reachable,
    )

    manifest = load_manifest(_managed_center_proxy_dir(home))
    if manifest is None:
        return ManagedApprovalCenterStatus(state="down", pid=None, port=None)
    if manifest_is_reachable(manifest):
        return ManagedApprovalCenterStatus(
            state="running",
            pid=manifest.pid,
            port=manifest.port,
        )
    return ManagedApprovalCenterStatus(
        state="stale",
        pid=manifest.pid,
        port=manifest.port,
    )


def clear_managed_approval_center_manifest(home: Path) -> bool:
    from agentveil_mcp_proxy.approval.persistent import manifest_path

    path = manifest_path(_managed_center_proxy_dir(home))
    try:
        path.unlink(missing_ok=True)
    except OSError:
        return False
    return True


def _managed_center_config_path(home: Path) -> Path:
    return _managed_center_proxy_dir(home) / "config.json"


def _managed_center_startup_log_path(home: Path) -> Path:
    return _managed_center_proxy_dir(home) / "approval-center.startup.log"


def _clear_managed_center_startup_log(home: Path) -> None:
    try:
        _managed_center_startup_log_path(home).unlink(missing_ok=True)
    except OSError:
        pass


def _command_name(command: str) -> str:
    if "\\" in command:
        return command.rsplit("\\", 1)[-1]
    return Path(command).name


def _is_python_command_name(name: str) -> bool:
    lowered = name.lower()
    if lowered in {"python", "python.exe", "python3", "python3.exe"}:
        return True
    if not lowered.startswith("python3."):
        return False
    version = lowered[len("python3.") :]
    if version.endswith(".exe"):
        version = version[:-4]
    if not version:
        return False
    for part in version.split("."):
        if not part.isdigit():
            return False
    return True


def _proxy_cli_argv(proxy_command: str, subcommand: list[str]) -> list[str]:
    if _is_python_command_name(_command_name(proxy_command)):
        script = (
            "from agentveil_mcp_proxy.cli import main; "
            f"raise SystemExit(main({subcommand!r}))"
        )
        return [proxy_command, "-c", script]
    resolved = shutil.which(proxy_command)
    if resolved:
        return [resolved, *subcommand]
    return [proxy_command, *subcommand]


_MANAGED_CENTER_ENV_KEYS = (
    "PATH",
    "USER",
    "LOGNAME",
    "SHELL",
    "HOME",
    "LANG",
    "LC_ALL",
    "LC_CTYPE",
    "LC_MESSAGES",
    "LC_NUMERIC",
    "LC_TIME",
    "TMPDIR",
    "TEMP",
    "TMP",
    "TZ",
    "TERM",
    "SYSTEMROOT",
    "WINDIR",
    "COMSPEC",
    "PATHEXT",
    "PYTHONHOME",
    "VIRTUAL_ENV",
)

_BLOCKED_MANAGED_CENTER_ENV_KEYS = frozenset({
    "OPENAI_API_KEY",
    "DEEPSEEK_API_KEY",
    "AWS_SECRET_ACCESS_KEY",
})

_BLOCKED_MANAGED_CENTER_ENV_SUFFIXES = ("_TOKEN", "_SECRET", "_PASSWORD")


def _managed_center_env_key_is_blocked(key: str) -> bool:
    upper = key.upper()
    if upper in _BLOCKED_MANAGED_CENTER_ENV_KEYS:
        return True
    return any(upper.endswith(suffix) for suffix in _BLOCKED_MANAGED_CENTER_ENV_SUFFIXES)


def _managed_center_pythonpath(*, parent_env: Mapping[str, str]) -> str:
    package_root_path = Path(__file__).resolve().parents[2]
    source_roots = [str(package_root_path)]
    repo_root = package_root_path.parents[1]
    if (repo_root / "agentveil").is_dir():
        source_roots.append(str(repo_root))
    existing = parent_env.get("PYTHONPATH")
    if existing:
        paths = existing.split(os.pathsep)
        prefix = [path for path in source_roots if path not in paths]
        if prefix:
            return os.pathsep.join([*prefix, existing])
        return existing
    return os.pathsep.join(source_roots)


def _proxy_cli_child_env(
    *,
    passphrase_file: Path | None = None,
    parent_env: Mapping[str, str] | None = None,
) -> dict[str, str]:
    """Return a bounded env for managed Approval Center child processes."""

    from agentveil_mcp_proxy.identity import PASSPHRASE_ENV

    source = dict(os.environ if parent_env is None else parent_env)
    env: dict[str, str] = {}
    for key in _MANAGED_CENTER_ENV_KEYS:
        if _managed_center_env_key_is_blocked(key):
            continue
        value = source.get(key)
        if isinstance(value, str) and value:
            env[key] = value
    env["PYTHONPATH"] = _managed_center_pythonpath(parent_env=source)
    if passphrase_file is None:
        passphrase = source.get(PASSPHRASE_ENV)
        if isinstance(passphrase, str) and passphrase:
            env[PASSPHRASE_ENV] = passphrase
    return env


def _path_match_tokens(path: Path) -> tuple[str, ...]:
    """Return path spellings that may appear in ``ps`` output on macOS."""

    tokens: set[str] = set()
    raw = Path(path)
    tokens.add(str(raw))
    try:
        resolved = str(raw.resolve())
        tokens.add(resolved)
        real = os.path.realpath(str(raw))
        tokens.add(real)
    except OSError:
        pass
    for value in list(tokens):
        if value.startswith("/private/"):
            tokens.add(value[len("/private") :])
        elif value.startswith("/"):
            tokens.add(f"/private{value}")
    for value in list(tokens):
        tokens.add(value.replace("\\", "/"))
        tokens.add(value.replace("/", "\\"))
    return tuple(token for token in tokens if token)


def _cmdline_contains_path(cmd: str, path: Path) -> bool:
    cmd_variants = {cmd, cmd.replace("\\", "/"), cmd.replace("/", "\\")}
    return any(
        token in cmd_variant
        for cmd_variant in cmd_variants
        for token in _path_match_tokens(path)
    )


def _read_process_command(pid: int) -> str:
    """Return the best-effort full command line for a PID."""

    if os.name != "posix":
        return ""
    for argv in (
        ["ps", "-ww", "-p", str(pid), "-o", "args="],
        ["ps", "-p", str(pid), "-o", "args="],
        ["ps", "-p", str(pid), "-o", "command="],
    ):
        try:
            result = subprocess.run(
                argv,
                capture_output=True,
                text=True,
                check=False,
                timeout=2.0,
            )
        except (OSError, subprocess.TimeoutExpired):
            continue
        cmd = result.stdout.strip()
        if result.returncode == 0 and cmd:
            return cmd
    return ""


def _process_cmdline_matches_managed_center(home: Path, pid: int) -> bool:
    """Best-effort cmdline proof that pid is our approval-center serve process."""

    cmd = _read_process_command(pid)
    if not cmd:
        return False
    lowered = cmd.lower()
    if "approval-center" not in lowered or "serve" not in lowered:
        return False
    if not _cmdline_contains_path(cmd, home):
        return False
    config = _managed_center_config_path(home)
    if _cmdline_contains_path(cmd, config):
        return True
    config_suffix = "/mcp-proxy/config.json"
    return any(f"{token}{config_suffix}" in cmd for token in _path_match_tokens(home))


def scan_cmdline_proven_managed_center_pids(home: Path) -> tuple[int, ...]:
    """Return PIDs whose cmdline proves they serve this project's Approval Center."""

    if os.name != "posix":
        return ()
    try:
        result = subprocess.run(
            ["ps", "ax", "-o", "pid=,command="],
            capture_output=True,
            text=True,
            check=False,
            timeout=5.0,
        )
    except (OSError, subprocess.TimeoutExpired):
        return ()
    if result.returncode != 0:
        return ()
    matched: set[int] = set()
    for line in result.stdout.splitlines():
        stripped = line.strip()
        if not stripped:
            continue
        pid_text, _, cmd = stripped.partition(" ")
        if not pid_text.isdigit() or not cmd:
            continue
        pid = int(pid_text)
        if _process_cmdline_matches_managed_center(home, pid):
            matched.add(pid)
    return tuple(sorted(matched))


def managed_center_cmdline_owns_pid(home: Path, pid: int) -> bool:
    return _managed_process_is_active(pid) and _process_cmdline_matches_managed_center(
        home, pid
    )


def managed_center_owns_pid(home: Path, manifest: Any) -> bool:
    """Return True only when pid is provably this project's managed Approval Center."""

    from agentveil_mcp_proxy.approval.persistent import manifest_is_reachable

    pid = manifest.pid
    if pid is None or not _managed_process_is_active(pid):
        return False
    if manifest_is_reachable(manifest):
        return True
    return _process_cmdline_matches_managed_center(home, pid)


def _managed_process_is_active(pid: int | None) -> bool:
    """Return True when pid refers to a live, non-zombie process."""

    from agentveil_mcp_proxy.approval.persistent import is_process_alive

    if not is_process_alive(pid):
        return False
    if pid is None or os.name != "posix":
        return True
    try:
        result = subprocess.run(
            ["ps", "-p", str(pid), "-o", "stat="],
            capture_output=True,
            text=True,
            check=False,
        )
    except OSError:
        return True
    if result.returncode != 0:
        return False
    state = result.stdout.strip()
    return bool(state) and "Z" not in state


def terminate_managed_approval_center_pid(
    pid: int,
    *,
    grace_seconds: float = _MANAGED_CENTER_STOP_GRACE_SECONDS,
) -> bool:
    if not _managed_process_is_active(pid):
        return True
    try:
        os.kill(pid, signal.SIGTERM)
    except ProcessLookupError:
        return True
    except PermissionError:
        return False

    deadline = time.monotonic() + grace_seconds
    while time.monotonic() < deadline:
        if not _managed_process_is_active(pid):
            return True
        time.sleep(0.05)

    try:
        os.kill(pid, signal.SIGKILL)
    except ProcessLookupError:
        return True
    except PermissionError:
        return False

    kill_deadline = time.monotonic() + 1.0
    while time.monotonic() < kill_deadline:
        if not _managed_process_is_active(pid):
            return True
        time.sleep(0.05)
    return not _managed_process_is_active(pid)


def stop_managed_approval_center(
    home: Path,
    *,
    require_healthy: bool = False,
) -> dict[str, Any]:
    """Stop only the Approval Center PID recorded in this project's manifest."""

    from agentveil_mcp_proxy.approval.persistent import load_manifest, manifest_is_reachable

    proxy_dir = _managed_center_proxy_dir(home)
    manifest = load_manifest(proxy_dir)
    if manifest is None or manifest.pid is None:
        clear_managed_approval_center_manifest(home)
        _clear_managed_center_startup_log(home)
        return {
            "stopped": False,
            "pid": None,
            "reason": "no managed approval-center manifest",
        }

    pid = manifest.pid
    healthy = manifest_is_reachable(manifest)
    owns = managed_center_owns_pid(home, manifest)
    unowned_reason = (
        "manifest pid is not a proven AgentVeil Approval Center; "
        "manifest cleared without kill"
    )

    if require_healthy:
        if not healthy:
            if _managed_process_is_active(pid):
                if managed_center_cmdline_owns_pid(home, pid):
                    stopped = terminate_managed_approval_center_pid(pid)
                    clear_managed_approval_center_manifest(home)
                    _clear_managed_center_startup_log(home)
                    if stopped:
                        reason = "managed approval-center stopped"
                    else:
                        reason = "managed approval-center stop did not confirm pid exit"
                    return {"stopped": stopped, "pid": pid, "reason": reason}
                return {
                    "stopped": False,
                    "pid": pid,
                    "reason": (
                        "manifest pid is not a healthy AgentVeil Approval Center; not stopped"
                    ),
                }
            clear_managed_approval_center_manifest(home)
            _clear_managed_center_startup_log(home)
            return {
                "stopped": False,
                "pid": pid,
                "reason": "managed pid not alive; manifest cleared",
            }
        if not owns:
            clear_managed_approval_center_manifest(home)
            _clear_managed_center_startup_log(home)
            return {"stopped": False, "pid": pid, "reason": unowned_reason}
    elif _managed_process_is_active(pid) and not owns:
        clear_managed_approval_center_manifest(home)
        _clear_managed_center_startup_log(home)
        return {"stopped": False, "pid": pid, "reason": unowned_reason}

    stopped = False
    if _managed_process_is_active(pid):
        stopped = terminate_managed_approval_center_pid(pid)
    clear_managed_approval_center_manifest(home)
    _clear_managed_center_startup_log(home)
    if stopped:
        reason = "managed approval-center stopped"
    elif not _managed_process_is_active(pid):
        reason = "managed pid already exited; manifest cleared"
    else:
        reason = "managed approval-center stop did not confirm pid exit"
    return {"stopped": stopped, "pid": pid, "reason": reason}


def prepare_stale_managed_approval_center(home: Path) -> dict[str, Any]:
    """Remove a stale managed center before starting a replacement."""

    from agentveil_mcp_proxy.approval.persistent import load_manifest

    status = inspect_managed_approval_center(home)
    if status.state != "stale":
        return {"prepared": False, "reason": "approval center is not stale", "pid": status.pid}

    manifest = load_manifest(_managed_center_proxy_dir(home))
    if manifest is None or manifest.pid is None:
        clear_managed_approval_center_manifest(home)
        _clear_managed_center_startup_log(home)
        return {
            "prepared": True,
            "stopped": False,
            "pid": None,
            "reason": "stale manifest cleared",
        }

    pid = manifest.pid
    if _managed_process_is_active(pid) and (
        managed_center_cmdline_owns_pid(home, pid)
        or managed_center_owns_pid(home, manifest)
    ):
        stopped = terminate_managed_approval_center_pid(pid)
        if not stopped and managed_center_cmdline_owns_pid(home, pid):
            stopped = terminate_managed_approval_center_pid(pid, grace_seconds=0.25)
        clear_managed_approval_center_manifest(home)
        _clear_managed_center_startup_log(home)
        if stopped:
            reason = "stale managed approval-center stopped"
        else:
            reason = "stale manifest cleared; managed stop did not confirm pid exit"
        return {"prepared": True, "stopped": stopped, "pid": pid, "reason": reason}

    clear_managed_approval_center_manifest(home)
    _clear_managed_center_startup_log(home)
    return {
        "prepared": True,
        "stopped": False,
        "pid": pid,
        "reason": "stale manifest cleared without killing unowned pid",
    }


def spawn_managed_approval_center_process(
    *,
    proxy_command: str,
    home: Path,
    passphrase_file: Path | None = None,
) -> subprocess.Popen[bytes]:
    """Spawn ``approval-center serve`` as a detached background process."""

    config_path = _managed_center_config_path(home)
    startup_log = _managed_center_startup_log_path(home)
    startup_log.parent.mkdir(parents=True, exist_ok=True)
    log_handle = open(startup_log, "ab")
    devnull = subprocess.DEVNULL
    kwargs: dict[str, Any] = {
        "stdin": devnull,
        "stdout": log_handle,
        "stderr": log_handle,
        "close_fds": True,
        "env": _proxy_cli_child_env(passphrase_file=passphrase_file),
    }
    if os.name == "posix":
        kwargs["start_new_session"] = True
    subcommand = [
        "approval-center",
        "serve",
        "--home",
        str(home),
        "--config",
        str(config_path),
        "--port",
        "0",
    ]
    if passphrase_file is not None:
        subcommand.extend(["--passphrase-file", str(passphrase_file)])
    try:
        return subprocess.Popen(  # noqa: S603 - args constructed from validated paths
            _proxy_cli_argv(proxy_command, subcommand),
            **kwargs,
        )
    finally:
        log_handle.close()


def ensure_managed_approval_center_for_cli(
    *,
    home: Path,
    proxy_command: str,
    passphrase_file: Path | None = None,
) -> ManagedApprovalCenterEnsureResult:
    """Idempotently ensure a healthy project-local Approval Center for setup flows."""

    def spawn() -> subprocess.Popen[bytes]:
        return spawn_managed_approval_center_process(
            proxy_command=proxy_command,
            home=home,
            passphrase_file=passphrase_file,
        )

    return ensure_managed_approval_center_running(home=home, spawn=spawn)


def ensure_managed_approval_center_running(
    *,
    home: Path,
    spawn: Callable[[], Any],
    wait_for_running: Callable[[Path, float], ManagedApprovalCenterStatus] | None = None,
) -> ManagedApprovalCenterEnsureResult:
    """Reuse a healthy managed center or replace stale state before spawning."""

    initial = inspect_managed_approval_center(home)
    if initial.state == "running":
        return ManagedApprovalCenterEnsureResult(
            status=initial,
            started=False,
            reused=True,
            restarted=False,
            reason="approval center already running",
        )

    restarted = initial.state == "stale"
    if restarted:
        prepare_stale_managed_approval_center(home)

    try:
        spawned = spawn()
    except (OSError, ValueError) as exc:
        return ManagedApprovalCenterEnsureResult(
            status=initial,
            started=False,
            reused=False,
            restarted=False,
            reason=f"could not spawn approval center: {exc.__class__.__name__}",
        )

    waiter = wait_for_running or _default_wait_for_managed_center_running
    deadline = time.monotonic() + _MANAGED_CENTER_START_TIMEOUT_SECONDS
    final = waiter(home, deadline)
    if final.state == "running":
        _clear_managed_center_startup_log(home)
        return ManagedApprovalCenterEnsureResult(
            status=final,
            started=True,
            reused=False,
            restarted=restarted,
            reason="approval center restarted" if restarted else "approval center started",
        )
    if isinstance(spawned, subprocess.Popen):
        returncode = spawned.poll()
        if returncode is not None:
            return ManagedApprovalCenterEnsureResult(
                status=final,
                started=False,
                reused=False,
                restarted=False,
                reason=(
                    "approval center process exited before becoming healthy "
                    f"(exit {returncode})"
                ),
            )
    return ManagedApprovalCenterEnsureResult(
        status=final,
        started=False,
        reused=False,
        restarted=False,
        reason="approval center did not become healthy within the start timeout",
    )


def _default_wait_for_managed_center_running(
    home: Path,
    deadline: float,
) -> ManagedApprovalCenterStatus:
    last = inspect_managed_approval_center(home)
    while time.monotonic() < deadline:
        last = inspect_managed_approval_center(home)
        if last.state == "running":
            return last
        time.sleep(_MANAGED_CENTER_POLL_INTERVAL_SECONDS)
    return last


__all__ = [
    "APPROVAL_DECISION_DENIED_BODY",
    "APPROVAL_DECISION_RECORDED_BODY",
    "APPROVAL_POST_APPROVE_HUMAN_BODY",
    "APPROVAL_POST_APPROVE_PROOF_BODY",
    "APPROVAL_POST_APPROVE_PROOF_PROMPT",
    "APPROVAL_LOCAL_URL_WARNING",
    "INTERNAL_REGISTER_TOKEN_HEADER",
    "ManagedApprovalCenterEnsureResult",
    "ManagedApprovalCenterStatus",
    "ApprovalServer",
    "ApprovalServerDecision",
    "ApprovalServerError",
    "ApprovalServerGone",
    "SECURITY_HEADERS",
    "approval_content_security_policy",
    "approval_security_headers",
    "clear_managed_approval_center_manifest",
    "ensure_managed_approval_center_for_cli",
    "ensure_managed_approval_center_running",
    "inspect_managed_approval_center",
    "managed_center_cmdline_owns_pid",
    "managed_center_owns_pid",
    "prepare_stale_managed_approval_center",
    "scan_cmdline_proven_managed_center_pids",
    "spawn_managed_approval_center_process",
    "render_local_proof_block",
    "render_local_proof_prompt_block",
    "stop_managed_approval_center",
    "terminate_managed_approval_center_pid",
    "TERMINAL_ALREADY_DECIDED",
    "TERMINAL_ALREADY_DECIDED_APPROVE",
    "TERMINAL_ALREADY_DECIDED_DENY",
    "TERMINAL_APPROVAL_EXPIRED",
    "TerminalApprovalSnapshot",
]
