user-engine/src/user_engine/web.py

"""Dependency-free WSGI transport for the user-engine portal.

Authentication is deliberately delegated to KeyCape (or another OIDC-aware
edge).  The application accepts claims only when the edge presents a shared
authentication marker configured at process start.  This keeps passwords,
MFA material, provider administration credentials, and browser sessions out
of user-engine.
"""

from __future__ import annotations

from dataclasses import asdict, is_dataclass, replace
from enum import Enum
from html import escape
import hashlib
import hmac
import json
import re
import secrets
from collections import deque
from threading import Lock
from time import monotonic
from typing import Any, Callable, Iterable, Mapping
from urllib.parse import parse_qs, urlencode, urlsplit

from user_engine.domain import (
    AccountStatus,
    Actor,
    FactorVerification,
    FamilyMemberSpec,
    IdentityFactorType,
    PrincipalType,
)
from user_engine.errors import AuthorizationDenied, ConflictError, NotFoundError, ValidationError
from user_engine.oidc import OIDCClient, cookie_value
from user_engine.ports import (
    IdentityProvisioningPort,
    ProvisioningRequest,
    RegistrationVerificationPort,
    RegistrationVerificationRequest,
    TenantManagementPort,
)
from user_engine.service import PLATFORM_TENANT, UserEngineService

StartResponse = Callable[[str, list[tuple[str, str]]], Any]


def _jsonable(value: Any) -> Any:
    if is_dataclass(value):
        return {key: _jsonable(item) for key, item in asdict(value).items()}
    if isinstance(value, Enum):
        return value.value
    if isinstance(value, Mapping):
        return {str(key): _jsonable(item) for key, item in value.items()}
    if isinstance(value, (tuple, list)):
        return [_jsonable(item) for item in value]
    if hasattr(value, "isoformat"):
        return value.isoformat()
    return value


class PortalApplication:
    """Small, auditable HTTP adapter over :class:`UserEngineService`."""

    def __init__(
        self,
        service: UserEngineService,
        *,
        trusted_proxy_secret: str,
        login_url: str,
        public_registration: bool = True,
        oidc_client: OIDCClient | None = None,
        provisioning: IdentityProvisioningPort | None = None,
        tenant_management: TenantManagementPort | None = None,
        outbox_delivery: Callable[[Any], None] | None = None,
        registration_verification: RegistrationVerificationPort | None = None,
        registration_clients: tuple[str, ...] = (),
        registration_tenants: tuple[str, ...] = (),
        registration_oidc_issuer: str = "",
        registration_password_setup_origins: tuple[str, ...] = (),
        registration_rate_limit: int = 10,
        registration_rate_window_seconds: int = 60,
    ) -> None:
        if len(trusted_proxy_secret) < 24:
            raise ValueError("trusted proxy secret must contain at least 24 characters")
        self.service = service
        self.trusted_proxy_secret = trusted_proxy_secret
        self.login_url = login_url
        self.public_registration = public_registration
        self.oidc_client = oidc_client
        self.provisioning = provisioning
        self.tenant_management = tenant_management
        self.outbox_delivery = outbox_delivery
        self.registration_verification = registration_verification
        self.registration_clients = frozenset(registration_clients)
        self.registration_tenants = frozenset(registration_tenants)
        self.registration_oidc_issuer = registration_oidc_issuer.rstrip("/")
        self.registration_password_setup_origins = frozenset(
            origin.rstrip("/") for origin in registration_password_setup_origins
        )
        if registration_rate_limit < 1 or registration_rate_window_seconds < 1:
            raise ValueError("registration rate limit and window must be positive")
        self.registration_rate_limit = registration_rate_limit
        self.registration_rate_window_seconds = registration_rate_window_seconds
        self._registration_attempts: dict[str, deque[float]] = {}
        self._registration_attempts_lock = Lock()

    def __call__(self, environ: Mapping[str, Any], start_response: StartResponse) -> Iterable[bytes]:
        correlation_id = environ.get("HTTP_X_REQUEST_ID") or f"corr_{secrets.token_hex(12)}"
        try:
            return self._dispatch(environ, start_response, str(correlation_id))
        except ConflictError as exc:
            return self._error(start_response, "409 Conflict", "conflict", str(exc), correlation_id)
        except (ValidationError, ValueError) as exc:
            return self._error(start_response, "400 Bad Request", "invalid_request", str(exc), correlation_id)
        except RuntimeError:
            return self._error(
                start_response,
                "502 Bad Gateway",
                "provisioning_unavailable",
                "Identity provisioning is temporarily unavailable.",
                correlation_id,
            )
        except AuthorizationDenied:
            return self._error(start_response, "403 Forbidden", "access_denied", "Access denied.", correlation_id)
        except NotFoundError:
            return self._error(start_response, "404 Not Found", "not_found", "Resource not found.", correlation_id)
        except (json.JSONDecodeError, UnicodeDecodeError):
            return self._error(start_response, "400 Bad Request", "invalid_json", "Malformed request body.", correlation_id)

    def _dispatch(self, environ: Mapping[str, Any], start_response: StartResponse, correlation_id: str) -> Iterable[bytes]:
        method = str(environ.get("REQUEST_METHOD", "GET")).upper()
        path = str(environ.get("PATH_INFO", "/")).rstrip("/") or "/"
        if method == "POST" and path in {
            "/register", "/registration/verify", "/registration/resume",
            "/api/v1/public/registrations",
            "/api/v1/public/registrations/verify",
            "/api/v1/public/registrations/resume",
        } and not self._accept_registration_attempt(environ):
            return self._error(
                start_response, "429 Too Many Requests", "rate_limited",
                "Too many registration attempts. Try again later.", correlation_id,
            )
        if path == "/healthz":
            return self._json(start_response, "200 OK", _jsonable(self.service.health()), correlation_id)
        if path == "/readyz":
            report = self.service.readiness()
            return self._json(start_response, "200 OK" if report.ready else "503 Service Unavailable", _jsonable(report), correlation_id)
        if path == "/metrics":
            supplied = str(environ.get("HTTP_X_USER_ENGINE_PROXY_SECRET", ""))
            if not secrets.compare_digest(supplied, self.trusted_proxy_secret):
                raise AuthorizationDenied("metrics require the trusted workload marker")
            return self._metrics(start_response, correlation_id)
        if path in {"/login", "/oidc/start"}:
            query = parse_qs(str(environ.get("QUERY_STRING", "")))
            tenant_hint = query.get("tenant_hint", [None])[0]
            if tenant_hint is not None and not str(tenant_hint).startswith("tenant:"):
                raise ValidationError("tenant_hint must be a tenant identifier")
            location = (
                self.oidc_client.begin(tenant_hint=str(tenant_hint) if tenant_hint else None)
                if self.oidc_client else self.login_url
            )
            start_response("303 See Other", [("Location", location), *self._security_headers(correlation_id)])
            return [b""]
        if path == "/oidc/callback":
            if self.oidc_client is None:
                raise NotFoundError("OIDC login is not configured")
            query = parse_qs(str(environ.get("QUERY_STRING", "")))
            if query.get("error"):
                raise AuthorizationDenied("OIDC login failed")
            session_id = self.oidc_client.complete(
                code=query.get("code", [""])[0],
                state=query.get("state", [""])[0],
            )
            headers = [
                ("Location", "/"),
                ("Set-Cookie", f"ue_session={session_id}; Path=/; HttpOnly; Secure; SameSite=Lax; Max-Age=3600"),
                *self._security_headers(correlation_id),
            ]
            start_response("303 See Other", headers)
            return [b""]
        if path == "/logout" and method == "POST":
            session_id = cookie_value(str(environ.get("HTTP_COOKIE", "")), "ue_session")
            if session_id and self.oidc_client:
                self.oidc_client.logout(session_id)
            start_response(
                "303 See Other",
                [("Location", "/"), ("Set-Cookie", "ue_session=; Path=/; HttpOnly; Secure; SameSite=Lax; Max-Age=0"), *self._security_headers(correlation_id)],
            )
            return [b""]
        if path == "/" and method == "GET":
            actor = self._optional_actor(environ)
            return self._html(start_response, self._home(actor), correlation_id)

        if path == "/register" and method == "GET":
            if not self.public_registration or self.registration_verification is None:
                raise NotFoundError("public registration is unavailable")
            token = secrets.token_urlsafe(32)
            return self._html(
                start_response, self._registration_form(token), correlation_id,
                extra_headers=[("Set-Cookie", self._registration_csrf_cookie(token))],
            )
        if path == "/register" and method == "POST":
            body = self._form_body(environ)
            self._require_registration_csrf(environ, str(body.get("csrf_token", "")))
            return self._start_public_registration(
                environ, start_response, correlation_id, body=body, browser=True
            )
        if path == "/registration/verify" and method == "GET":
            query = parse_qs(str(environ.get("QUERY_STRING", "")))
            handle = str(query.get("handle", [""])[0])
            if len(handle) < 16:
                raise ValidationError("verification handle is invalid")
            token = secrets.token_urlsafe(32)
            return self._html(
                start_response, self._registration_verification_form(token, handle),
                correlation_id,
                extra_headers=[("Set-Cookie", self._registration_csrf_cookie(token))],
            )
        if path == "/registration/verify" and method == "POST":
            body = self._form_body(environ)
            self._require_registration_csrf(environ, str(body.get("csrf_token", "")))
            return self._verify_public_registration(
                environ, start_response, correlation_id, body=body, browser=True
            )
        if path == "/registration/resume" and method == "POST":
            body = self._form_body(environ)
            self._require_registration_csrf(environ, str(body.get("csrf_token", "")))
            return self._resume_public_registration(
                environ, start_response, correlation_id, body=body, browser=True
            )

        if path == "/api/v1/public/registrations" and method == "POST":
            return self._start_public_registration(
                environ, start_response, correlation_id
            )
        if path == "/api/v1/public/registrations/verify" and method == "POST":
            return self._verify_public_registration(
                environ, start_response, correlation_id
            )
        if path == "/api/v1/public/registrations/resume" and method == "POST":
            return self._resume_public_registration(
                environ, start_response, correlation_id
            )

        actor = self._actor(environ)
        if path == "/api/v1/me" and method == "GET":
            return self._json(start_response, "200 OK", _jsonable(self.service.me(self._claims(environ), correlation_id=correlation_id)), correlation_id)
        if path == "/api/v1/me/profile" and method == "PATCH":
            self._idempotency_key(environ)
            self.service.me(self._claims(environ), correlation_id=correlation_id)
            body = self._body(environ)
            updated = self.service.update_self_service_profile(
                actor, display_name=str(body.get("display_name", "")),
                consent_accepted=bool(body.get("consent_accepted", False)),
                consent_version=str(body.get("consent_version") or "portal-terms-v1"),
                correlation_id=correlation_id,
            )
            return self._json(start_response, "200 OK", _jsonable(updated), correlation_id)
        if path == "/onboarding" and method == "GET":
            session = self.service.me(self._claims(environ), correlation_id=correlation_id)
            memberships = self.service.store.memberships_for_user(session.user.user_id)
            journeys = self.service.store.onboarding_journeys_for_user(session.user.user_id)
            query = parse_qs(str(environ.get("QUERY_STRING", "")))
            selected_tenant = query.get("tenant", [session.actor.tenant])[0]
            allowed_tenants = {item.tenant for item in memberships} | {session.actor.tenant}
            if selected_tenant not in allowed_tenants:
                raise AuthorizationDenied("tenant selection is not a membership")
            return self._html(
                start_response,
                self._onboarding(
                    session, memberships, journeys, str(selected_tenant),
                    self._csrf_token(environ),
                ),
                correlation_id,
            )
        if path == "/onboarding/profile" and method == "POST":
            body = self._form_body(environ)
            self._require_csrf(environ, str(body.get("csrf_token", "")))
            self.service.me(self._claims(environ), correlation_id=correlation_id)
            self.service.update_self_service_profile(
                actor, display_name=str(body.get("display_name", "")),
                consent_accepted=body.get("consent_accepted") == "yes",
                consent_version="portal-terms-v1",
                correlation_id=correlation_id,
            )
            return self._redirect(start_response, "/onboarding", correlation_id)
        if path.startswith("/onboarding/") and "/steps/" in path and path.endswith("/complete") and method == "POST":
            body = self._form_body(environ)
            self._require_csrf(environ, str(body.get("csrf_token", "")))
            parts = path.split("/")
            journey_id, step_key = parts[2], parts[4]
            session = self.service.me(self._claims(environ), correlation_id=correlation_id)
            journey = self.service.store.onboarding_journey(journey_id)
            if journey is None or journey.user_id != session.user.user_id:
                raise NotFoundError("onboarding journey not found")
            step = next((item for item in journey.steps if item.step_key == step_key), None)
            if step is None:
                raise NotFoundError("onboarding step not found")
            if step.subsystem != "user-engine" or step.handoff is not None:
                raise AuthorizationDenied("subsystem-owned steps require their handoff")
            self.service.complete_onboarding_step(
                actor, journey_id, step_key, correlation_id=correlation_id
            )
            return self._redirect(start_response, "/onboarding", correlation_id)
        if path.startswith("/invitations/") and method == "GET":
            invitation = self.service.store.family_invitation(path.split("/")[2])
            if invitation is None:
                raise NotFoundError("invitation not found")
            self.service.resolve_tenant_context(actor, invitation.tenant)
            return self._html(
                start_response,
                self._invitation_acceptance(invitation, self._csrf_token(environ)),
                correlation_id,
            )
        if path.startswith("/invitations/") and method == "POST":
            body = self._form_body(environ)
            self._require_csrf(environ, str(body.get("csrf_token", "")))
            self.service.accept_family_invitation(
                self._claims(environ), path.split("/")[2], correlation_id=correlation_id
            )
            return self._redirect(start_response, "/onboarding", correlation_id)
        if path == "/api/v1/platform/tenants" and method == "POST":
            self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
            if self.tenant_management is None:
                raise ValidationError("tenant management is unavailable")
            idempotency_key = self._idempotency_key(environ)
            body = self._body(environ)
            tenant = str(body.get("tenant") or "")
            if not tenant.startswith("tenant:") or tenant == PLATFORM_TENANT:
                raise ValidationError("a non-platform tenant identifier is required")
            result = self.tenant_management.create_tenant(
                tenant=tenant,
                display_name=str(body.get("display_name") or tenant),
                idempotency_key=idempotency_key,
                correlation_id=correlation_id,
            )
            admin = body.get("first_admin")
            bootstrap = None
            if admin is not None:
                if not isinstance(admin, Mapping):
                    raise ValidationError("first_admin must be an object")
                user = self.service.create_user(
                    actor, display_name=admin.get("display_name"),
                    primary_email=admin.get("primary_email"),
                    correlation_id=correlation_id,
                )
                account = self.service.set_tenant_account_status(
                    actor, user.user_id, AccountStatus.INVITED,
                    tenant=tenant, correlation_id=correlation_id,
                )
                membership = self.service.add_membership(
                    actor, user.user_id, tenant=tenant, scope_type="tenant",
                    scope_id=tenant, kind="tenant-admin",
                    correlation_id=correlation_id,
                )
                bootstrap = {"user": user, "tenant_account": account, "membership": membership}
            return self._json(start_response, "201 Created", {
                "tenant": _jsonable(result), "first_admin": _jsonable(bootstrap),
            }, correlation_id)
        if path == "/api/v1/platform/outbox/deliver" and method == "POST":
            self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
            if self.outbox_delivery is None:
                raise ValidationError("outbox delivery is unavailable")
            body = self._body(environ)
            events = self.service.deliver_outbox(
                actor, self.outbox_delivery,
                worker_id=str(body.get("worker_id") or "portal-operator"),
                max_attempts=int(body.get("max_attempts") or 3),
                correlation_id=correlation_id,
            )
            return self._json(start_response, "200 OK", {"items": _jsonable(events)}, correlation_id)
        if path.startswith("/api/v1/platform/outbox/") and path.endswith("/replay") and method == "POST":
            self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
            event = self.service.replay_outbox(
                actor, path.split("/")[5], correlation_id=correlation_id
            )
            return self._json(start_response, "200 OK", _jsonable(event), correlation_id)
        if path.startswith("/api/v1/platform/tenants/") and path.endswith("/recover") and method == "POST":
            self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
            if self.provisioning is None:
                raise ValidationError("identity provisioning is unavailable")
            parts = path.split("/")
            tenant, user_id = parts[5], parts[7]
            idempotency_key = self._idempotency_key(environ)
            user = self.service.store.user(user_id)
            if user is None:
                raise NotFoundError("user not found")
            request = ProvisioningRequest(
                user_id=user_id, tenant=tenant, primary_email=user.primary_email,
                display_name=user.display_name, idempotency_key=idempotency_key,
                correlation_id=correlation_id,
                roles=tuple(item.kind for item in self.service.store.memberships_for_user(user_id, tenant=tenant)),
            )
            identity = next(iter(self.service.store.identities_for_user(user_id)), None)
            if identity is None:
                provisioned = self.provisioning.provision(request)
                self.service.link_identity(
                    actor, user_id, issuer="urn:netkingdom:directory",
                    subject=provisioned.external_subject, provider=provisioned.provider,
                    correlation_id=correlation_id,
                )
                recovery = {"status": provisioned.status, "changed": ("identity",)}
            else:
                reconciled = self.provisioning.reconcile(
                    request, external_subject=identity.subject, desired_status="active"
                )
                recovery = {"status": reconciled.status, "drift": reconciled.drift, "changed": reconciled.changed}
            account = self.service.set_tenant_account_status(
                actor, user_id, AccountStatus.ACTIVE, tenant=tenant,
                correlation_id=correlation_id,
            )
            return self._json(start_response, "200 OK", {
                "recovery": _jsonable(recovery), "tenant_account": _jsonable(account),
            }, correlation_id)
        if path.startswith("/api/v1/invitations/") and path.endswith("/claim") and method == "POST":
            invitation_id = path.split("/")[4]
            accepted = self.service.accept_family_invitation(
                self._claims(environ), invitation_id, correlation_id=correlation_id
            )
            return self._json(start_response, "200 OK", _jsonable(accepted), correlation_id)
        if path.startswith("/api/v1/onboarding/") and "/steps/" in path and path.endswith("/complete") and method == "POST":
            self._idempotency_key(environ)
            parts = path.split("/")
            journey_id, step_key = parts[4], parts[6]
            session = self.service.me(self._claims(environ), correlation_id=correlation_id)
            journey = self.service.store.onboarding_journey(journey_id)
            if journey is None or journey.user_id != session.user.user_id:
                raise NotFoundError("onboarding journey not found")
            step = next((item for item in journey.steps if item.step_key == step_key), None)
            if step is None:
                raise NotFoundError("onboarding step not found")
            if step.subsystem != "user-engine" or step.handoff is not None:
                raise AuthorizationDenied("subsystem-owned steps require their handoff")
            updated = self.service.complete_onboarding_step(
                actor, journey_id, step_key, correlation_id=correlation_id
            )
            return self._json(start_response, "200 OK", _jsonable(updated), correlation_id)
        if path == "/api/v1/registrations" and method == "POST":
            if not self.public_registration:
                raise AuthorizationDenied("public registration disabled")
            body = self._body(environ)
            session = self.service.start_registration(
                actor,
                tenant=body.get("tenant"),
                correlation_id=correlation_id,
            )
            return self._json(start_response, "201 Created", _jsonable(session), correlation_id)
        if path.startswith("/api/v1/registrations/") and path.endswith("/complete") and method == "POST":
            registration_id = path.split("/")[4]
            body = self._body(environ)
            result = self.service.complete_registration(
                actor,
                registration_id,
                display_name=body.get("display_name"),
                primary_email=body.get("primary_email"),
                correlation_id=correlation_id,
            )
            return self._json(start_response, "200 OK", _jsonable(result), correlation_id)
        if path.startswith("/api/v1/tenants/") and path.endswith("/users") and method == "GET":
            tenant = path.split("/")[4]
            self.service.resolve_tenant_context(actor, tenant)
            memberships = self.service.store.memberships_for_tenant(tenant)
            offset, limit = self._page(environ)
            items = memberships[offset : offset + limit]
            payload = {"items": _jsonable(items), "offset": offset, "limit": limit, "total": len(memberships)}
            return self._json(start_response, "200 OK", payload, correlation_id)
        if path.startswith("/api/v1/tenants/") and path.endswith("/users") and method == "POST":
            tenant = path.split("/")[4]
            self.service.resolve_tenant_context(actor, tenant)
            body = self._body(environ)
            user = self.service.create_user(
                actor,
                display_name=body.get("display_name"),
                primary_email=body.get("primary_email"),
                correlation_id=correlation_id,
            )
            # Platform operators may create an identity for a tenant other than
            # their own.  Ensure the lifecycle record follows the requested
            # tenant instead of only retaining the actor tenant created by the
            # generic domain operation.
            tenant_account = self.service.set_tenant_account_status(
                actor,
                user.user_id,
                AccountStatus.ACTIVE,
                tenant=tenant,
                correlation_id=correlation_id,
            )
            membership = self.service.add_membership(
                actor,
                user.user_id,
                tenant=tenant,
                scope_type="tenant",
                scope_id=tenant,
                kind=str(body.get("role", "user")),
                correlation_id=correlation_id,
            )
            return self._json(start_response, "201 Created", {
                "user": _jsonable(user),
                "tenant_account": _jsonable(tenant_account),
                "membership": _jsonable(membership),
                "provisioning_status": "pending",
            }, correlation_id)
        if path.startswith("/api/v1/tenants/") and path.endswith("/invitations"):
            tenant = path.split("/")[4]
            self.service.resolve_tenant_context(actor, tenant)
            if method == "GET":
                items = self.service.store.family_invitations_for_tenant(tenant)
                return self._json(start_response, "200 OK", {"items": _jsonable(items)}, correlation_id)
            if method == "POST":
                body = self._body(environ)
                invited = self.service.invite_family_member(
                    actor,
                    tenant=tenant,
                    family_scope_id=str(body.get("scope_id") or tenant),
                    application_id=str(body.get("application_id") or "app.user-portal"),
                    member=FamilyMemberSpec(
                        primary_email=str(body.get("primary_email") or ""),
                        display_name=body.get("display_name"),
                        role=str(body.get("role") or "user"),
                    ),
                    correlation_id=correlation_id,
                )
                return self._json(start_response, "201 Created", _jsonable(invited), correlation_id)
        if path.startswith("/api/v1/tenants/") and "/invitations/" in path and method == "POST":
            parts = path.split("/")
            tenant, invitation_id, action = parts[4], parts[6], parts[7]
            self.service.resolve_tenant_context(actor, tenant)
            expected = self._expected_version(environ)
            if action == "resend":
                value = self.service.resend_family_invitation(
                    actor, invitation_id, correlation_id=correlation_id,
                    expected_version=expected,
                )
            elif action == "expire":
                value = self.service.revoke_family_invitation(
                    actor, invitation_id, correlation_id=correlation_id,
                    expected_version=expected,
                )
            else:
                raise NotFoundError("invitation action not found")
            return self._json(start_response, "200 OK", _jsonable(value), correlation_id)
        if path.startswith("/api/v1/tenants/") and path.endswith("/provision") and method == "POST":
            if self.provisioning is None:
                raise ValidationError("identity provisioning is unavailable")
            parts = path.split("/")
            tenant, user_id = parts[4], parts[6]
            self.service.resolve_tenant_context(actor, tenant)
            user = self.service.store.user(user_id)
            if user is None:
                raise NotFoundError("user not found")
            idempotency_key = str(environ.get("HTTP_IDEMPOTENCY_KEY", ""))
            if len(idempotency_key) < 16:
                raise ValidationError("Idempotency-Key must contain at least 16 characters")
            result = self.provisioning.provision(ProvisioningRequest(
                user_id=user.user_id,
                tenant=tenant,
                primary_email=user.primary_email,
                display_name=user.display_name,
                idempotency_key=idempotency_key,
                correlation_id=correlation_id,
                roles=tuple(
                    membership.kind
                    for membership in self.service.store.memberships_for_user(
                        user.user_id, tenant=tenant
                    )
                ),
            ))
            identity = self.service.link_identity(
                actor,
                user.user_id,
                issuer="urn:netkingdom:directory",
                subject=result.external_subject,
                provider=result.provider,
                correlation_id=correlation_id,
            )
            return self._json(start_response, "200 OK", {
                "provisioning": _jsonable(result),
                "identity": _jsonable(identity),
            }, correlation_id)
        if path.startswith("/api/v1/tenants/") and "/users/" in path and method == "PATCH":
            parts = path.split("/")
            tenant, user_id = parts[4], parts[6]
            body = self._body(environ)
            status = AccountStatus(str(body["status"]))
            idempotency_key = str(environ.get("HTTP_IDEMPOTENCY_KEY", ""))
            if len(idempotency_key) < 16:
                raise ValidationError("Idempotency-Key must contain at least 16 characters")
            result = self._change_status(
                actor, tenant, user_id, status,
                idempotency_key=idempotency_key,
                correlation_id=correlation_id,
            )
            return self._json(start_response, "200 OK", _jsonable(result), correlation_id)
        if path.startswith("/api/v1/tenants/") and "/users/" in path and method == "DELETE":
            if self.provisioning is None:
                raise ValidationError("identity provisioning is unavailable")
            parts = path.split("/")
            tenant, user_id = parts[4], parts[6]
            self.service.resolve_tenant_context(actor, tenant)
            idempotency_key = str(environ.get("HTTP_IDEMPOTENCY_KEY", ""))
            if len(idempotency_key) < 16:
                raise ValidationError("Idempotency-Key must contain at least 16 characters")
            identity = next(iter(self.service.store.identities_for_user(user_id)), None)
            if identity is not None:
                self.provisioning.deprovision(
                    external_subject=identity.subject,
                    idempotency_key=idempotency_key,
                    correlation_id=correlation_id,
                )
            account = self.service.set_tenant_account_status(
                actor, user_id, AccountStatus.DISABLED,
                tenant=tenant, correlation_id=correlation_id,
            )
            return self._json(start_response, "200 OK", {
                "status": "removed", "tenant_account": _jsonable(account),
                "provider_identity_removed": identity is not None,
            }, correlation_id)
        if path == "/platform" and method == "GET":
            self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
            return self._html(
                start_response, self._platform(self._csrf_token(environ)), correlation_id
            )
        if path == "/platform/tenants" and method == "POST":
            self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
            if self.tenant_management is None:
                raise ValidationError("tenant management is unavailable")
            body = self._form_body(environ)
            self._require_csrf(environ, str(body.get("csrf_token", "")))
            tenant = str(body.get("tenant", ""))
            if not tenant.startswith("tenant:") or tenant == PLATFORM_TENANT:
                raise ValidationError("a non-platform tenant identifier is required")
            result = self.tenant_management.create_tenant(
                tenant=tenant, display_name=str(body.get("display_name") or tenant),
                idempotency_key=f"portal-tenant-{tenant}", correlation_id=correlation_id,
            )
            email = str(body.get("admin_email", ""))
            if email:
                user = self.service.create_user(
                    actor, display_name=body.get("admin_display_name"),
                    primary_email=email, correlation_id=correlation_id,
                )
                self.service.set_tenant_account_status(
                    actor, user.user_id, AccountStatus.INVITED,
                    tenant=tenant, correlation_id=correlation_id,
                )
                self.service.add_membership(
                    actor, user.user_id, tenant=tenant, scope_type="tenant",
                    scope_id=tenant, kind="tenant-admin", correlation_id=correlation_id,
                )
            return self._html(
                start_response,
                self._platform_result(result, tenant, bool(email)), correlation_id,
            )
        if path.startswith("/admin/") and method == "GET":
            tenant = path.split("/")[2]
            self.service.resolve_tenant_context(actor, tenant)
            memberships = self.service.store.memberships_for_tenant(tenant)
            invitations = self.service.store.family_invitations_for_tenant(tenant)
            diagnostics = self.service.tenant_diagnostics(
                actor, tenant=tenant, correlation_id=correlation_id
            )
            return self._html(
                start_response,
                self._admin(
                    tenant, memberships, invitations, diagnostics,
                    "platform-operator" in actor.roles, self._csrf_token(environ),
                ),
                correlation_id,
            )
        if path.startswith("/admin/") and method == "POST":
            parts = path.split("/")
            tenant = parts[2]
            self.service.resolve_tenant_context(actor, tenant)
            body = self._form_body(environ)
            self._require_csrf(environ, str(body.get("csrf_token", "")))
            if len(parts) == 4 and parts[3] == "users":
                user = self.service.create_user(
                    actor,
                    display_name=body.get("display_name"),
                    primary_email=body.get("primary_email"),
                    correlation_id=correlation_id,
                )
                self.service.set_tenant_account_status(
                    actor, user.user_id, AccountStatus.ACTIVE,
                    tenant=tenant, correlation_id=correlation_id,
                )
                self.service.add_membership(
                    actor, user.user_id, tenant=tenant, scope_type="tenant",
                    scope_id=tenant, kind=str(body.get("role", "user")),
                    correlation_id=correlation_id,
                )
                return self._redirect(start_response, f"/admin/{tenant}", correlation_id)
            if len(parts) == 4 and parts[3] == "invitations":
                self.service.invite_family_member(
                    actor, tenant=tenant, family_scope_id=tenant,
                    application_id="app.user-portal",
                    member=FamilyMemberSpec(
                        primary_email=str(body.get("primary_email", "")),
                        display_name=body.get("display_name"),
                        role=str(body.get("role", "user")),
                    ), correlation_id=correlation_id,
                )
                return self._redirect(start_response, f"/admin/{tenant}", correlation_id)
            if len(parts) == 6 and parts[3] == "invitations":
                invitation_id, action = parts[4], parts[5]
                version = int(body.get("version", "0"))
                if action == "resend":
                    self.service.resend_family_invitation(
                        actor, invitation_id, expected_version=version,
                        correlation_id=correlation_id,
                    )
                elif action == "expire":
                    self.service.revoke_family_invitation(
                        actor, invitation_id, expected_version=version,
                        correlation_id=correlation_id,
                    )
                else:
                    raise NotFoundError("invitation action not found")
                return self._redirect(start_response, f"/admin/{tenant}", correlation_id)
            if len(parts) == 6 and parts[3] == "users" and parts[5] == "provision":
                if self.provisioning is None:
                    raise ValidationError("identity provisioning is unavailable")
                user_id = parts[4]
                user = self.service.store.user(user_id)
                if user is None:
                    raise NotFoundError("user not found")
                result = self.provisioning.provision(ProvisioningRequest(
                    user_id=user.user_id,
                    tenant=tenant,
                    primary_email=user.primary_email,
                    display_name=user.display_name,
                    idempotency_key=f"portal-{user.user_id}-{tenant}",
                    correlation_id=correlation_id,
                    roles=tuple(
                        item.kind for item in self.service.store.memberships_for_user(
                            user.user_id, tenant=tenant
                        )
                    ),
                ))
                self.service.link_identity(
                    actor, user.user_id, issuer="urn:netkingdom:directory",
                    subject=result.external_subject, provider=result.provider,
                    correlation_id=correlation_id,
                )
                if result.password_setup_url:
                    return self._html(
                        start_response,
                        self._password_setup_handoff(
                            result.password_setup_url, tenant
                        ),
                        correlation_id,
                    )
                query = urlencode({"provisioned": user.user_id, "status": result.status})
                return self._redirect(start_response, f"/admin/{tenant}?{query}", correlation_id)
            if len(parts) == 6 and parts[3] == "users" and parts[5] == "status":
                status = AccountStatus(str(body.get("status", "")))
                if status not in {AccountStatus.ACTIVE, AccountStatus.SUSPENDED}:
                    raise ValidationError("browser lifecycle supports active or suspended")
                self._change_status(
                    actor, tenant, parts[4], status,
                    idempotency_key=f"portal-status-{parts[4]}-{status.value}",
                    correlation_id=correlation_id,
                )
                return self._redirect(start_response, f"/admin/{tenant}", correlation_id)
            if len(parts) == 6 and parts[3] == "users" and parts[5] == "remove":
                if self.provisioning is None:
                    raise ValidationError("identity provisioning is unavailable")
                user_id = parts[4]
                identity = next(iter(self.service.store.identities_for_user(user_id)), None)
                if identity is not None:
                    self.provisioning.deprovision(
                        external_subject=identity.subject,
                        idempotency_key=f"portal-remove-{tenant}-{user_id}",
                        correlation_id=correlation_id,
                    )
                self.service.set_tenant_account_status(
                    actor, user_id, AccountStatus.DISABLED,
                    tenant=tenant, correlation_id=correlation_id,
                )
                return self._redirect(start_response, f"/admin/{tenant}", correlation_id)
            if len(parts) == 6 and parts[3] == "users" and parts[5] == "recover":
                self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
                if self.provisioning is None:
                    raise ValidationError("identity provisioning is unavailable")
                user_id = parts[4]
                user = self.service.store.user(user_id)
                if user is None:
                    raise NotFoundError("user not found")
                request = ProvisioningRequest(
                    user_id=user_id, tenant=tenant, primary_email=user.primary_email,
                    display_name=user.display_name,
                    idempotency_key=f"portal-recover-{tenant}-{user_id}",
                    correlation_id=correlation_id,
                    roles=tuple(item.kind for item in self.service.store.memberships_for_user(user_id, tenant=tenant)),
                )
                identity = next(iter(self.service.store.identities_for_user(user_id)), None)
                if identity is None:
                    result = self.provisioning.provision(request)
                    self.service.link_identity(
                        actor, user_id, issuer="urn:netkingdom:directory",
                        subject=result.external_subject, provider=result.provider,
                        correlation_id=correlation_id,
                    )
                else:
                    self.provisioning.reconcile(
                        request, external_subject=identity.subject, desired_status="active"
                    )
                self.service.set_tenant_account_status(
                    actor, user_id, AccountStatus.ACTIVE,
                    tenant=tenant, correlation_id=correlation_id,
                )
                return self._redirect(start_response, f"/admin/{tenant}?recovered={user_id}", correlation_id)
        return self._error(start_response, "404 Not Found", "not_found", "Resource not found.", correlation_id)

    def _change_status(
        self,
        actor: Any,
        tenant: str,
        user_id: str,
        status: AccountStatus,
        *,
        idempotency_key: str,
        correlation_id: str,
    ) -> Any:
        if self.provisioning is None:
            raise ValidationError("identity provisioning is unavailable")
        self.service.resolve_tenant_context(actor, tenant)
        identity = next(
            (
                item for item in self.service.store.identities_for_user(user_id)
                if item.provider == "netkingdom-lldap"
            ),
            None,
        )
        if identity is None:
            raise ValidationError("user has no managed login identity")
        if status == AccountStatus.SUSPENDED:
            self.provisioning.suspend(
                external_subject=identity.subject,
                idempotency_key=idempotency_key,
                correlation_id=correlation_id,
            )
        elif status == AccountStatus.ACTIVE:
            self.provisioning.reactivate(
                external_subject=identity.subject,
                idempotency_key=idempotency_key,
                correlation_id=correlation_id,
            )
        else:
            raise ValidationError("provider lifecycle supports active or suspended")
        return self.service.set_tenant_account_status(
            actor, user_id, status, tenant=tenant, correlation_id=correlation_id
        )

    def _claims(self, environ: Mapping[str, Any]) -> Mapping[str, Any]:
        if self.oidc_client is not None:
            session_id = cookie_value(str(environ.get("HTTP_COOKIE", "")), "ue_session")
            if session_id:
                claims = self.oidc_client.claims(session_id)
                if claims is not None:
                    return claims
        marker = str(environ.get("HTTP_X_USER_ENGINE_PROXY_SECRET", ""))
        if not secrets.compare_digest(marker, self.trusted_proxy_secret):
            raise AuthorizationDenied("untrusted identity source")
        raw = environ.get("HTTP_X_VERIFIED_OIDC_CLAIMS")
        if not raw:
            raise AuthorizationDenied("verified claims required")
        claims = json.loads(str(raw))
        if not isinstance(claims, dict):
            raise AuthorizationDenied("verified claims must be an object")
        return claims

    def _actor(self, environ: Mapping[str, Any]) -> Any:
        return self.service.identity_adapter.normalize(self._claims(environ))

    def _start_public_registration(
        self,
        environ: Mapping[str, Any],
        start_response: StartResponse,
        correlation_id: str,
        *, body: Mapping[str, Any] | None = None, browser: bool = False,
    ) -> Iterable[bytes]:
        if not self.public_registration or self.registration_verification is None:
            raise NotFoundError("public registration is unavailable")
        body = body if body is not None else self._body(environ)
        username = self._registration_username(body.get("username"))
        email = self._registration_email(body.get("email"))
        display_name = str(body.get("display_name") or "").strip() or None
        if display_name is not None and len(display_name) > 200:
            raise ValidationError("display_name is too long")
        client_id = str(body.get("client_id") or "")
        tenant = str(body.get("tenant") or "")
        if client_id not in self.registration_clients:
            raise ValidationError("client_id is not eligible for registration")
        if tenant not in self.registration_tenants:
            raise ValidationError("tenant is not eligible for registration")

        applicant_subject = f"applicant_{secrets.token_hex(16)}"
        actor = Actor(
            issuer="urn:netkingdom:public-registration",
            subject=applicant_subject,
            tenant=tenant,
            principal_type=PrincipalType.HUMAN,
            audience=("user-engine",),
            roles=("registration-applicant",),
            authorized_party=client_id,
            preferred_username=username,
        )
        session = self.service.start_registration(
            actor,
            tenant=tenant,
            correlation_id=correlation_id,
            applicant_username=username,
            client_id=client_id,
        )
        self.registration_verification.request(
            RegistrationVerificationRequest(
                registration_id=session.registration_id,
                normalized_email=email,
                preferred_username=username,
                client_id=client_id,
                tenant=tenant,
                correlation_id=correlation_id,
                display_name=display_name,
            )
        )
        if browser:
            return self._html(
                start_response,
                self._page_html(
                    "Check your email",
                    "<h1>Check your email.</h1>"
                    "<p>If the address can be registered, a verification link is on its way. "
                    "The link expires after 30 minutes.</p>",
                ),
                correlation_id,
            )
        return self._json(
            start_response,
            "202 Accepted",
            {"status": "verification_requested"},
            correlation_id,
        )

    def _verify_public_registration(
        self,
        environ: Mapping[str, Any],
        start_response: StartResponse,
        correlation_id: str,
        *, body: Mapping[str, Any] | None = None, browser: bool = False,
    ) -> Iterable[bytes]:
        if not self.public_registration or self.registration_verification is None:
            raise NotFoundError("public registration is unavailable")
        body = body if body is not None else self._body(environ)
        evidence = self.registration_verification.consume(str(body.get("handle") or ""))
        session = self.service.store.registration_session(evidence.registration_id)
        if session is None:
            raise NotFoundError("registration not found")
        if (
            evidence.client_id != session.client_id
            or evidence.tenant != session.tenant
            or evidence.preferred_username != session.applicant_username
        ):
            raise AuthorizationDenied("verification evidence does not match registration")
        actor = Actor(
            issuer="urn:netkingdom:public-registration",
            subject=str(session.started_by_subject),
            tenant=session.tenant,
            principal_type=PrincipalType.HUMAN,
            audience=("user-engine",),
            roles=("registration-applicant",),
            authorized_party=session.client_id,
            preferred_username=session.applicant_username,
        )
        updated = self.service.attach_registration_factor(
            actor,
            session.registration_id,
            FactorVerification(
                factor_type=IdentityFactorType.EMAIL,
                normalized_value=evidence.normalized_email,
                verification_id=evidence.verification_id,
                source_system=evidence.source_system,
                assurance=dict(evidence.assurance),
            ),
            correlation_id=correlation_id,
        )
        resume_handle = secrets.token_urlsafe(32)
        updated = replace(
            updated,
            provisioning_resume_hash=hashlib.sha256(resume_handle.encode()).hexdigest(),
        )
        self.service.store.save_registration_session(updated)
        if self.provisioning is None or not self.registration_oidc_issuer:
            raise RuntimeError("public registration provisioning is unavailable")
        completion = self.service.complete_registration(
            actor,
            updated.registration_id,
            display_name=evidence.display_name,
            primary_email=evidence.normalized_email,
            correlation_id=correlation_id,
        )
        try:
            return self._provision_public_registration(
                start_response, actor, completion.user, completion.session,
                evidence.normalized_email, evidence.display_name,
                evidence.preferred_username, correlation_id, browser=browser,
            )
        except RuntimeError:
            if browser:
                token = secrets.token_urlsafe(32)
                return self._html(
                    start_response,
                    self._registration_resume_form(
                        token, updated.registration_id, resume_handle
                    ),
                    correlation_id,
                    extra_headers=[("Set-Cookie", self._registration_csrf_cookie(token))],
                )
            return self._json(start_response, "202 Accepted", {
                "status": "provisioning_pending",
                "registration_id": updated.registration_id,
                "resume_handle": resume_handle,
            }, correlation_id)

    def _resume_public_registration(
        self, environ, start_response, correlation_id, *,
        body: Mapping[str, Any] | None = None, browser: bool = False,
    ):
        if not self.public_registration:
            raise NotFoundError("public registration is unavailable")
        body = body if body is not None else self._body(environ)
        session = self.service.store.registration_session(str(body.get("registration_id") or ""))
        handle = str(body.get("resume_handle") or "")
        digest = hashlib.sha256(handle.encode()).hexdigest()
        if (
            session is None or session.status.value != "completed"
            or not session.provisioning_resume_hash
            or not hmac.compare_digest(digest, session.provisioning_resume_hash)
        ):
            raise AuthorizationDenied("registration resume is invalid")
        factors = self.service.store.factors_for_registration(session.registration_id)
        email_factor = next((item for item in factors if item.factor_type == IdentityFactorType.EMAIL), None)
        if email_factor is None or not session.user_id:
            raise ValidationError("registration is not recoverable")
        actor = Actor(
            issuer="urn:netkingdom:public-registration", subject=str(session.started_by_subject),
            tenant=session.tenant, principal_type=PrincipalType.HUMAN,
            audience=("user-engine",), roles=("registration-applicant",),
            authorized_party=session.client_id, preferred_username=session.applicant_username,
        )
        user = self.service.store.user(session.user_id)
        if user is None:
            raise ValidationError("registration user is unavailable")
        return self._provision_public_registration(
            start_response, actor, user, session, email_factor.normalized_value,
            user.display_name, str(session.applicant_username), correlation_id,
            browser=browser,
        )

    def _provision_public_registration(
        self, start_response, actor, user, session, email, display_name,
        preferred_username, correlation_id,
        *, browser: bool = False,
    ):
        provisioned = self.provisioning.provision(
            ProvisioningRequest(
                user_id=user.user_id,
                tenant=session.tenant,
                primary_email=email,
                display_name=display_name,
                preferred_username=preferred_username,
                idempotency_key=f"public-registration-{session.registration_id}",
                correlation_id=correlation_id,
                roles=("user",),
            )
        )
        self.service.link_identity(
            actor,
            user.user_id,
            issuer=self.registration_oidc_issuer,
            subject=provisioned.external_subject,
            provider=provisioned.provider,
            correlation_id=correlation_id,
        )
        self.service.store.save_registration_session(
            replace(session, provisioning_resume_hash=None)
        )
        if provisioned.password_setup_url:
            self._validate_registration_handoff(provisioned.password_setup_url)
            if browser:
                return self._html(
                    start_response,
                    self._page_html(
                        "Create your password",
                        "<h1>Your identity is ready.</h1>"
                        "<p>Continue to the protected identity service to create your password.</p>"
                        f'<p><a class="button" rel="noreferrer" href="{escape(provisioned.password_setup_url)}">Create password</a></p>',
                    ),
                    correlation_id,
                )
            return self._redirect(
                start_response, provisioned.password_setup_url, correlation_id
            )
        return self._json(
            start_response,
            "200 OK",
            {"status": "identity_ready", "login_required": True},
            correlation_id,
        )

    def _validate_registration_handoff(self, setup_url: str) -> None:
        parsed = urlsplit(setup_url)
        origin = f"{parsed.scheme}://{parsed.netloc}"
        if (
            parsed.scheme != "https"
            or not parsed.netloc
            or origin not in self.registration_password_setup_origins
        ):
            raise ValidationError("password setup handoff is not allow-listed")

    @staticmethod
    def _registration_username(value: Any) -> str:
        username = str(value or "").strip().lower()
        if not re.fullmatch(r"[a-z][a-z0-9._-]{2,31}", username):
            raise ValidationError("username is invalid")
        if username in {"admin", "administrator", "platform-root", "root", "system"}:
            raise ValidationError("username is reserved")
        return username

    @staticmethod
    def _registration_email(value: Any) -> str:
        email = str(value or "").strip().lower()
        if len(email) > 254 or not re.fullmatch(r"[^@\s]+@[^@\s]+\.[^@\s]+", email):
            raise ValidationError("email is invalid")
        return email

    def _optional_actor(self, environ: Mapping[str, Any]) -> Any | None:
        try:
            return self._actor(environ)
        except (AuthorizationDenied, json.JSONDecodeError, ValidationError):
            return None

    @staticmethod
    def _body(environ: Mapping[str, Any]) -> Mapping[str, Any]:
        length = min(int(environ.get("CONTENT_LENGTH") or 0), 65536)
        payload = environ["wsgi.input"].read(length) if length else b"{}"
        value = json.loads(payload.decode("utf-8"))
        if not isinstance(value, dict):
            raise ValidationError("request body must be an object")
        return value

    @staticmethod
    def _form_body(environ: Mapping[str, Any]) -> Mapping[str, str]:
        content_type = str(environ.get("CONTENT_TYPE", "")).partition(";")[0]
        if content_type != "application/x-www-form-urlencoded":
            raise ValidationError("form content type is required")
        length = min(int(environ.get("CONTENT_LENGTH") or 0), 65536)
        payload = environ["wsgi.input"].read(length).decode("utf-8")
        return {key: values[0] for key, values in parse_qs(payload).items()}

    def _csrf_token(self, environ: Mapping[str, Any]) -> str:
        if self.oidc_client is None:
            raise AuthorizationDenied("browser session required")
        session_id = cookie_value(str(environ.get("HTTP_COOKIE", "")), "ue_session")
        token = self.oidc_client.csrf_token(session_id or "")
        if token is None:
            raise AuthorizationDenied("browser session required")
        return token

    def _require_csrf(self, environ: Mapping[str, Any], supplied: str) -> None:
        expected = self._csrf_token(environ)
        if not supplied or not secrets.compare_digest(supplied, expected):
            raise AuthorizationDenied("invalid CSRF token")

    @staticmethod
    def _registration_csrf_cookie(token: str) -> str:
        return (
            f"ue_registration_csrf={token}; Path=/; HttpOnly; Secure; "
            "SameSite=Strict; Max-Age=1800"
        )

    def _require_registration_csrf(
        self, environ: Mapping[str, Any], supplied: str
    ) -> None:
        expected = cookie_value(
            str(environ.get("HTTP_COOKIE", "")), "ue_registration_csrf"
        )
        if not supplied or not expected or not secrets.compare_digest(supplied, expected):
            raise AuthorizationDenied("invalid registration CSRF token")

    def _accept_registration_attempt(self, environ: Mapping[str, Any]) -> bool:
        """Apply a bounded per-source sliding-window limit to public writes.

        Only ``REMOTE_ADDR`` is trusted. Forwarded headers are deliberately
        ignored because the ingress must normalize the peer address before the
        request reaches this process.
        """
        source = str(environ.get("REMOTE_ADDR") or "unknown")[:128]
        now = monotonic()
        cutoff = now - self.registration_rate_window_seconds
        with self._registration_attempts_lock:
            attempts = self._registration_attempts.setdefault(source, deque())
            while attempts and attempts[0] <= cutoff:
                attempts.popleft()
            if len(attempts) >= self.registration_rate_limit:
                return False
            attempts.append(now)
            if len(self._registration_attempts) > 4096:
                for key in tuple(self._registration_attempts):
                    values = self._registration_attempts[key]
                    while values and values[0] <= cutoff:
                        values.popleft()
                    if not values:
                        del self._registration_attempts[key]
            return True

    @staticmethod
    def _page(environ: Mapping[str, Any]) -> tuple[int, int]:
        query = parse_qs(str(environ.get("QUERY_STRING", "")))
        offset = max(0, int(query.get("offset", ["0"])[0]))
        limit = max(1, min(100, int(query.get("limit", ["25"])[0])))
        return offset, limit

    @staticmethod
    def _expected_version(environ: Mapping[str, Any]) -> int:
        value = str(environ.get("HTTP_IF_MATCH", "")).strip().strip('"')
        if not value.isdigit():
            raise ValidationError("If-Match invitation version is required")
        return int(value)

    @staticmethod
    def _idempotency_key(environ: Mapping[str, Any]) -> str:
        value = str(environ.get("HTTP_IDEMPOTENCY_KEY", ""))
        if len(value) < 16:
            raise ValidationError("Idempotency-Key must contain at least 16 characters")
        return value

    def _home(self, actor: Any | None) -> str:
        identity = (
            f"<p>Signed in as <strong>{escape(actor.preferred_username)}</strong>.</p>"
            '<p><a class="button" href="/onboarding">Continue onboarding</a></p>'
            if actor is not None
            else (
                f'<p><a class="button" href="/login">Sign in with KeyCape</a></p>'
                + (
                    '<p>New here? <a href="/register">Create an account</a>.</p>'
                    if self.public_registration and self.registration_verification is not None
                    else ""
                )
            )
        )
        return self._page_html(
            "Identity & access",
            "<h1>Your account, on your terms.</h1>"
            "<p>Join a tenant, complete onboarding, and manage access without exposing credentials to applications.</p>"
            + identity,
        )

    def _registration_form(self, csrf_token: str) -> str:
        client_options = "".join(
            f'<option value="{escape(item)}">{escape(item)}</option>'
            for item in sorted(self.registration_clients)
        )
        tenant_options = "".join(
            f'<option value="{escape(item)}">{escape(item.removeprefix("tenant:"))}</option>'
            for item in sorted(self.registration_tenants)
        )
        return self._page_html(
            "Create account",
            f"""<h1>Create your account.</h1>
<p>We will verify your email before creating an identity.</p>
<form method="post" action="/register">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<label>Username <input name="username" required minlength="3" maxlength="32" pattern="[A-Za-z][A-Za-z0-9._-]+" autocomplete="username"></label>
<label>Email <input name="email" type="email" required autocomplete="email"></label>
<label>Display name <input name="display_name" maxlength="200" autocomplete="name"></label>
<label>Application <select name="client_id" required>{client_options}</select></label>
<label>Community <select name="tenant" required>{tenant_options}</select></label>
<button type="submit">Send verification email</button>
</form><p><a href="/login">Already have an account? Sign in</a></p>""",
        )

    def _registration_verification_form(self, csrf_token: str, handle: str) -> str:
        return self._page_html(
            "Verify email",
            f"""<h1>Verify your email.</h1>
<p>Confirm to finish creating your identity. This verification link can be used once.</p>
<form method="post" action="/registration/verify">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<input type="hidden" name="handle" value="{escape(handle)}">
<button type="submit">Verify and continue</button></form>""",
        )

    def _registration_resume_form(
        self, csrf_token: str, registration_id: str, resume_handle: str
    ) -> str:
        return self._page_html(
            "Finish account setup",
            f"""<h1>Your email is verified.</h1>
<p>The identity service is temporarily unavailable. Try again without repeating verification.</p>
<form method="post" action="/registration/resume">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<input type="hidden" name="registration_id" value="{escape(registration_id)}">
<input type="hidden" name="resume_handle" value="{escape(resume_handle)}">
<button type="submit">Try identity setup again</button></form>""",
        )
    def _admin(
        self, tenant: str, memberships: tuple[Any, ...],
        invitations: tuple[Any, ...], diagnostics: Any,
        platform_operator: bool, csrf_token: str,
    ) -> str:
        rows = "".join(
            self._admin_row(tenant, item, platform_operator, csrf_token)
            for item in memberships
        ) or '<tr><td colspan="6">No members yet.</td></tr>'
        invitation_rows = "".join(
            self._invitation_admin_row(tenant, item, csrf_token)
            for item in invitations
        ) or '<tr><td colspan="5">No invitations yet.</td></tr>'
        diagnostic_items = "".join(
            f"<li>{escape(item.replace('_', ' '))}</li>" for item in diagnostics.issues
        ) or "<li>No lifecycle gaps detected.</li>"
        return self._page_html(
            f"{tenant} users",
            f"""<h1>{escape(tenant)} users</h1>
<section><h2>Add a user</h2>
<form method="post" action="/admin/{escape(tenant)}/users">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<label>Display name <input name="display_name" required autocomplete="name"></label>
<label>Email <input name="primary_email" type="email" required autocomplete="email"></label>
<label>Role <select name="role"><option value="user">User</option><option value="tenant-admin">Tenant administrator</option></select></label>
<button type="submit">Add user</button></form></section>
<section><h2>Invite a user</h2>
<form method="post" action="/admin/{escape(tenant)}/invitations">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<label>Display name <input name="display_name" autocomplete="name"></label>
<label>Email <input name="primary_email" type="email" required autocomplete="email"></label>
<label>Role <select name="role"><option value="user">User</option><option value="tenant-admin">Tenant administrator</option></select></label>
<button type="submit">Send invitation</button></form></section>
<section><h2>Invitations</h2><table><thead><tr><th>Email</th><th>Role</th><th>Status</th><th>Expires</th><th>Action</th></tr></thead><tbody>{invitation_rows}</tbody></table></section>
<section aria-labelledby="lifecycle-gaps"><h2 id="lifecycle-gaps">Lifecycle diagnostics</h2><p>Diagnostics contain machine-readable gap categories only; credentials and factor evidence are never displayed.</p><ul>{diagnostic_items}</ul></section>
<section><h2>Members</h2><table><thead><tr><th>User</th><th>Email</th><th>Role</th><th>Status</th><th>Directory</th><th>Action</th></tr></thead><tbody>{rows}</tbody></table></section>""",
        )

    def _invitation_admin_row(self, tenant: str, invitation: Any, csrf_token: str) -> str:
        actions = ""
        if invitation.status.value == "pending":
            actions = f"""<form method="post" action="/admin/{escape(tenant)}/invitations/{escape(invitation.invitation_id)}/resend">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}"><input type="hidden" name="version" value="{invitation.version}"><button type="submit">Resend</button></form>
<form method="post" action="/admin/{escape(tenant)}/invitations/{escape(invitation.invitation_id)}/expire">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}"><input type="hidden" name="version" value="{invitation.version}"><button type="submit">Expire</button></form>"""
        expires = invitation.expires_at.isoformat() if invitation.expires_at else "—"
        return (
            f"<tr><td>{escape(invitation.primary_email)}</td><td>{escape(invitation.role)}</td>"
            f"<td>{escape(invitation.status.value)}</td><td>{escape(expires)}</td><td>{actions}</td></tr>"
        )

    def _admin_row(
        self, tenant: str, membership: Any,
        platform_operator: bool, csrf_token: str,
    ) -> str:
        user = self.service.store.user(membership.user_id)
        identities = self.service.store.identities_for_user(membership.user_id)
        directory = next(
            (item for item in identities if item.provider == "netkingdom-lldap"),
            None,
        )
        tenant_account = self.service.store.tenant_account(tenant, membership.user_id)
        status = tenant_account.status if tenant_account else AccountStatus.INVITED
        action = (
            f"""<span>Linked as {escape(directory.subject)}</span>
<form method="post" action="/admin/{escape(tenant)}/users/{escape(membership.user_id)}/provision">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<button type="submit">Create password setup link</button></form>
<form method="post" action="/admin/{escape(tenant)}/users/{escape(membership.user_id)}/status">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<input type="hidden" name="status" value="{'active' if status == AccountStatus.SUSPENDED else 'suspended'}">
<button type="submit">{'Reactivate' if status == AccountStatus.SUSPENDED else 'Suspend'}</button></form>"""
            if directory
            else f"""<form method="post" action="/admin/{escape(tenant)}/users/{escape(membership.user_id)}/provision">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<button type="submit">Create login</button></form>"""
        )
        action += f"""<form method="post" action="/admin/{escape(tenant)}/users/{escape(membership.user_id)}/remove">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<button type="submit">Remove account</button></form>"""
        if platform_operator:
            action += f"""<form method="post" action="/admin/{escape(tenant)}/users/{escape(membership.user_id)}/recover">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<button type="submit">Recover identity</button></form>"""
        return (
            f"<tr><td>{escape(user.display_name or membership.user_id) if user else escape(membership.user_id)}</td>"
            f"<td>{escape(user.primary_email or '') if user else ''}</td>"
            f"<td>{escape(membership.kind)}</td>"
            f"<td>{escape(status.value)}</td>"
            f"<td>{'linked' if directory else 'pending'}</td><td>{action}</td></tr>"
        )

    def _invitation_acceptance(self, invitation: Any, csrf_token: str) -> str:
        return self._page_html(
            "Accept invitation",
            f"""<h1>Join {escape(invitation.tenant)}</h1>
<p>You were invited as <strong>{escape(invitation.role)}</strong>. The invitation expires at {escape(invitation.expires_at.isoformat() if invitation.expires_at else 'the tenant policy deadline')}.</p>
<form method="post" action="/invitations/{escape(invitation.invitation_id)}">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<button type="submit">Accept invitation</button></form>
<p>Your password and MFA remain on the identity-provider surface.</p>""",
        )

    def _platform(self, csrf_token: str) -> str:
        return self._page_html(
            "Platform administration",
            f"""<h1>Platform administration</h1>
<section aria-labelledby="create-tenant"><h2 id="create-tenant">Create tenant</h2>
<form method="post" action="/platform/tenants">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<label>Tenant identifier <input name="tenant" required pattern="tenant:.+" placeholder="tenant:friendly:example"></label>
<label>Display name <input name="display_name" required></label>
<fieldset><legend>First administrator (optional)</legend>
<label>Name <input name="admin_display_name" autocomplete="name"></label>
<label>Email <input name="admin_email" type="email" autocomplete="email"></label></fieldset>
<button type="submit">Create tenant</button></form></section>""",
        )

    def _platform_result(self, result: Any, tenant: str, admin_prepared: bool) -> str:
        return self._page_html(
            "Tenant created",
            f"""<h1>Tenant {escape(result.status)}</h1>
<p><strong>{escape(tenant)}</strong> was processed by the tenant authority.</p>
<p>{'The first administrator is prepared and awaiting onboarding.' if admin_prepared else 'No first administrator was requested.'}</p>
<p><a class="button" href="/admin/{escape(tenant)}">Open tenant administration</a></p>
<p><a href="/platform">Return to platform administration</a></p>""",
        )

    def _onboarding(
        self, session: Any, memberships: tuple[Any, ...], journeys: tuple[Any, ...],
        selected_tenant: str, csrf_token: str,
    ) -> str:
        membership_items = "".join(
            f"<li><a href=\"/onboarding?{urlencode({'tenant': item.tenant})}\">{escape(item.tenant)}</a> — {escape(item.kind)}</li>"
            for item in memberships
        ) or "<li>No tenant memberships yet.</li>"
        journey_items = "".join(
            self._onboarding_journey_item(item, csrf_token) for item in journeys
        ) or "<li>No additional onboarding steps are required.</li>"
        verification = "Verified by your identity provider" if session.actor.assurance else "Verification pending"
        consent_checked = " checked" if session.user.consented_at else ""
        return self._page_html(
            "Onboarding",
            f"""<h1>Welcome, {escape(session.user.display_name or session.actor.preferred_username or session.user.user_id)}</h1>
<section aria-labelledby="verification"><h2 id="verification">Email and sign-in</h2><p>{escape(verification)}</p><p>Passwords and MFA are managed by your identity provider.</p></section>
<section aria-labelledby="profile"><h2 id="profile">Profile and consent</h2>
<form method="post" action="/onboarding/profile"><input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<label>Display name <input name="display_name" required maxlength="200" autocomplete="name" value="{escape(session.user.display_name or '')}"></label>
<label><input name="consent_accepted" type="checkbox" value="yes"{consent_checked}> I accept portal terms version 1</label>
<button type="submit">Save profile</button></form></section>
<section aria-labelledby="tenants"><h2 id="tenants">Tenant access</h2><p>Viewing <strong>{escape(selected_tenant)}</strong>.</p><ul>{membership_items}</ul>
<p><a href="/login?{urlencode({'tenant_hint': selected_tenant})}">Reauthenticate in this tenant</a> to change the authoritative login context.</p></section>
<section aria-labelledby="steps"><h2 id="steps">Onboarding progress</h2><ul>{journey_items}</ul></section>""",
        )

    @staticmethod
    def _onboarding_journey_item(journey: Any, csrf_token: str) -> str:
        steps = "".join(
            PortalApplication._onboarding_step_item(journey.journey_id, step, csrf_token)
            for step in journey.steps
        )
        return (
            f"<li><strong>{escape(journey.status.value)}</strong>"
            f"<ol>{steps}</ol></li>"
        )

    @staticmethod
    def _onboarding_step_item(journey_id: str, step: Any, csrf_token: str) -> str:
        action = ""
        if (
            step.status.value == "in_progress"
            and step.subsystem == "user-engine"
            and step.handoff is None
        ):
            action = f"""<form method="post" action="/onboarding/{escape(journey_id)}/steps/{escape(step.step_key)}/complete">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<button type="submit">Mark complete</button></form>"""
        elif step.handoff is not None or step.subsystem != "user-engine":
            action = "<span>Continue on the provider-owned surface; this page will resume after its callback.</span>"
        gap = f" <span>Support category: {escape(step.lifecycle_gap)}</span>" if step.lifecycle_gap else ""
        return (
            f"<li><span>{escape(step.title)} — {escape(step.status.value)}</span>{gap}{action}</li>"
        )

    def _password_setup_handoff(self, setup_url: str, tenant: str) -> str:
        if not setup_url.startswith("https://"):
            raise ValidationError("password setup handoff must use HTTPS")
        return self._page_html(
            "Password setup",
            "<h1>Login identity created</h1>"
            "<p>The password is handled only by the NetKingdom identity "
            "surface. This short-lived link is single use.</p>"
            f'<p><a class="button" rel="noreferrer" href="{escape(setup_url)}">'
            "Continue to password setup</a></p>"
            f'<p><a href="/admin/{escape(tenant)}">'
            "Return to tenant administration</a></p>",
        )

    def _redirect(self, start_response: StartResponse, location: str, correlation_id: str) -> list[bytes]:
        start_response("303 See Other", [("Location", location), *self._security_headers(correlation_id)])
        return [b""]

    @staticmethod
    def _page_html(title: str, body: str) -> str:
        return f"""<!doctype html><html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>{escape(title)} · Railiance</title><style>
:root{{--ink:#17201c;--paper:#f5f1e8;--accent:#195b47;--line:#c8c1b3}}
*{{box-sizing:border-box}}body{{margin:0;background:var(--paper);color:var(--ink);font:18px/1.55 system-ui,sans-serif}}
header,main{{max-width:68rem;margin:auto;padding:1.25rem}}header{{border-bottom:1px solid var(--line)}}
h1{{font:clamp(2.2rem,7vw,5.5rem)/.98 Georgia,serif;max-width:13ch}}a{{color:var(--accent)}}
.button{{display:inline-block;background:var(--accent);color:white;padding:.8rem 1.15rem;border-radius:.3rem;text-decoration:none}}
table{{width:100%;border-collapse:collapse;background:#fff}}th,td{{padding:.75rem;text-align:left;border-bottom:1px solid var(--line)}}
section{{margin:2rem 0}}form{{display:grid;gap:.8rem;max-width:42rem}}label{{display:grid;gap:.25rem}}
input,select,button{{font:inherit;padding:.65rem}}button{{background:var(--accent);color:white;border:0;border-radius:.3rem;cursor:pointer}}
a:focus-visible,input:focus-visible,select:focus-visible,button:focus-visible{{outline:3px solid #e59f24;outline-offset:3px}}@media(max-width:640px){{body{{font-size:16px}}table{{display:block;overflow-x:auto}}}}
</style></head><body><header><strong>Railiance identity</strong></header><main>{body}</main></body></html>"""

    def _html(
        self, start_response: StartResponse, body: str, correlation_id: str,
        *, extra_headers: list[tuple[str, str]] | None = None,
    ) -> list[bytes]:
        data = body.encode()
        start_response("200 OK", [
            ("Content-Type", "text/html; charset=utf-8"),
            ("Content-Length", str(len(data))),
            *(extra_headers or []),
            *self._security_headers(correlation_id),
        ])
        return [data]

    def _json(self, start_response: StartResponse, status: str, payload: Any, correlation_id: str) -> list[bytes]:
        data = json.dumps(payload, separators=(",", ":"), default=str).encode()
        start_response(status, [("Content-Type", "application/json"), ("Content-Length", str(len(data))), *self._security_headers(correlation_id)])
        return [data]

    def _metrics(
        self, start_response: StartResponse, correlation_id: str
    ) -> list[bytes]:
        report = self.service.readiness()
        counts = self.service.operability_snapshot().metrics
        lines = [
            "# HELP user_engine_ready Whether runtime dependencies are ready.",
            "# TYPE user_engine_ready gauge",
            f"user_engine_ready {1 if report.ready else 0}",
            "# HELP user_engine_records Durable logical record counts by kind.",
            "# TYPE user_engine_records gauge",
        ]
        for kind, count in sorted(counts.items()):
            safe_kind = "".join(
                character for character in str(kind)
                if character.isalnum() or character in "_-"
            )
            lines.append(f'user_engine_records{{kind="{safe_kind}"}} {int(count)}')
        data = ("\n".join(lines) + "\n").encode()
        start_response(
            "200 OK",
            [
                ("Content-Type", "text/plain; version=0.0.4; charset=utf-8"),
                ("Content-Length", str(len(data))),
                *self._security_headers(correlation_id),
            ],
        )
        return [data]

    def _error(self, start_response: StartResponse, status: str, code: str, message: str, correlation_id: str) -> list[bytes]:
        return self._json(start_response, status, {"error": {"code": code, "message": message, "correlation_id": correlation_id}}, correlation_id)

    @staticmethod
    def _security_headers(correlation_id: str) -> list[tuple[str, str]]:
        return [
            ("X-Request-ID", correlation_id),
            ("Cache-Control", "no-store"),
            ("X-Content-Type-Options", "nosniff"),
            ("Referrer-Policy", "no-referrer"),
            ("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'; form-action 'self'; frame-ancestors 'none'; base-uri 'none'"),
        ]