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 contextvars import ContextVar
from dataclasses import asdict, is_dataclass, replace
from enum import Enum
from html import escape
import time
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.error import URLError
from urllib.parse import parse_qs, quote, unquote, 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]

# Rendering state is scoped to one request, including concurrent WSGI requests.
_ACCOUNT_NAVIGATION: ContextVar[str] = ContextVar("account_navigation", default="")
_BROWSER_REQUEST: ContextVar[bool] = ContextVar("browser_request", default=False)


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,
        mfa_management_url: str = "",
        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
        if mfa_management_url:
            destination = urlsplit(mfa_management_url)
            if (destination.scheme != "https" or not destination.hostname
                    or destination.username or destination.password
                    or destination.query or destination.fragment
                    or any(c.isspace() for c in mfa_management_url)):
                raise ValueError("MFA management URL must be a fixed HTTPS destination without credentials or query")
        self.mfa_management_url = mfa_management_url
        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)}"
        navigation_token = _ACCOUNT_NAVIGATION.set("")
        browser_token = _BROWSER_REQUEST.set(
            "text/html" in str(environ.get("HTTP_ACCEPT", ""))
            and not str(environ.get("PATH_INFO", "/")).startswith("/api/")
            and str(environ.get("PATH_INFO", "/")) not in {"/healthz", "/readyz", "/metrics"}
        )
        try:
            if (not str(environ.get("PATH_INFO", "/")).startswith("/api/")
                    and str(environ.get("PATH_INFO", "/")) not in {"/healthz", "/readyz", "/metrics"}):
                self._set_account_navigation(environ, self._optional_actor(environ))
            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)
        finally:
            _ACCOUNT_NAVIGATION.reset(navigation_token)
            _BROWSER_REQUEST.reset(browser_token)

    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",
            "/registration/cancel",
            "/api/v1/public/registrations",
            "/api/v1/public/registrations/verify",
            "/api/v1/public/registrations/resume",
            "/api/v1/public/registrations/cancel",
        } 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", "")))
            try:
                if query.get("error"):
                    self.oidc_client.pending.pop(query.get("state", [""])[0], None)
                    raise ValueError("OIDC login failed")
                session_id = self.oidc_client.complete(
                    code=query.get("code", [""])[0],
                    state=query.get("state", [""])[0],
                )
            except (ValueError, URLError, OSError):
                return self._redirect(start_response, "/access-recovery", correlation_id)
            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 == "/access-recovery" and method == "GET":
            try:
                actor = self._optional_actor(environ)
            except AuthorizationDenied:
                actor = None
            self._set_account_navigation(environ, actor)
            identity = (
                f'<p>This portal is signed in as <strong>{escape(actor.preferred_username or actor.subject)}</strong>.</p>'
                '<p><a class="button" href="/onboarding">View my account and access</a></p>'
                if actor else '<p>You are not signed in to this portal. Sign in to verify your identity and access.</p>'
            )
            return self._html(start_response, self._page_html(
                "Sign-in help", '<h1>Sign-in could not be completed</h1>'
                '<p>Your account may not have access to the application, or sign-in may have been interrupted.</p>'
                + identity
                + '<p><a href="/security">Help with passwords and verification codes</a></p>'
                + (self._identity_switch_help() if actor is None else ""),
            ), correlation_id)
        if path == "/security" and method == "GET":
            return self._html(start_response, self._security_page(), correlation_id)
        if path == "/logged-out" and method == "GET":
            if self._optional_actor(environ) is not None:
                return self._redirect(start_response, "/", correlation_id)
            return self._html(start_response, self._page_html(
                "Not signed in",
                '<h1>You are not signed in to this portal.</h1>'
                '<p>Your shared NetKingdom sign-in may still be active. Signing in may reuse that account.</p>'
                + self._identity_switch_help(),
            ), correlation_id)
        if path == "/logout" and method == "GET":
            actor = self._optional_actor(environ)
            if actor is None:
                return self._redirect(start_response, "/logged-out", correlation_id)
            self._set_account_navigation(environ, actor)
            token = self._csrf_token(environ)
            return self._html(start_response, self._page_html(
                "Log out", '<h1>Log out of this portal?</h1>'
                '<p>This ends your portal session. Your shared NetKingdom sign-in stays active.</p>'
                '<form method="post" action="/logout">'
                f'<input type="hidden" name="csrf_token" value="{escape(token)}">'
                '<button type="submit">Log out of this portal</button>'
                '<button type="submit" name="scope" value="shared">Continue to NetKingdom sign-out</button></form>',
            ), correlation_id)
        if path == "/logout" and method == "POST":
            session_id = cookie_value(str(environ.get("HTTP_COOKIE", "")), "ue_session")
            body = self._form_body(environ)
            if session_id and self.oidc_client and self.oidc_client.claims(session_id) is not None:
                self._require_csrf(environ, str(body.get("csrf_token", "")))
                self.oidc_client.logout(session_id)
            start_response(
                "303 See Other",
                [("Location", self.oidc_client.issuer + "/account/logout" if body.get("scope") == "shared" and self.oidc_client else "/logged-out"), ("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)
            self._set_account_navigation(environ, actor)
            return self._html(start_response, self._home(actor), correlation_id)

        if path == "/register" and method == "GET":
            if self._optional_actor(environ) is not None:
                return self._redirect(start_response, "/onboarding", correlation_id)
            if not self.public_registration or self.registration_verification is None:
                raise NotFoundError("public registration is unavailable")
            token = secrets.token_urlsafe(32)
            idempotency_key = secrets.token_urlsafe(24)
            return self._html(
                start_response,
                self._registration_form(token, idempotency_key), 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 == "/registration/cancel" and method == "GET":
            query = parse_qs(str(environ.get("QUERY_STRING", "")))
            handle = str(query.get("handle", [""])[0])
            if len(handle) < 16:
                raise ValidationError("cancellation handle is invalid")
            token = secrets.token_urlsafe(32)
            return self._html(
                start_response, self._registration_cancel_form(token, handle),
                correlation_id,
                extra_headers=[("Set-Cookie", self._registration_csrf_cookie(token))],
            )
        if path == "/registration/cancel" and method == "POST":
            body = self._form_body(environ)
            self._require_registration_csrf(environ, str(body.get("csrf_token", "")))
            return self._cancel_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
            )
        if path == "/api/v1/public/registrations/cancel" and method == "POST":
            return self._cancel_public_registration(
                environ, start_response, correlation_id
            )

        actor = self._actor(environ)
        self._set_account_navigation(environ, actor)
        if path.startswith("/api/v1/tenants/"):
            self._require_tenant_admin(actor, path.split("/")[4])
        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)
            try:
                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,
                )
            except ValidationError:
                name = escape(str(body.get("display_name", ""))[:201])
                csrf = escape(self._csrf_token(environ))
                checked = " checked" if body.get("consent_accepted") == "yes" else ""
                page = self._page_html("Check your profile", f'''<h1>Check your profile</h1>
<p role="alert">Enter a display name of 1 to 200 characters. Your profile has not been saved.</p>
<form method="post" action="/onboarding/profile"><input type="hidden" name="csrf_token" value="{csrf}">
<label>Display name <input name="display_name" value="{name}" required maxlength="200" aria-invalid="true"></label>
<label><input type="checkbox" name="consent_accepted" value="yes"{checked}> I accept portal terms version 1</label>
<button type="submit">Save profile</button></form><p><a href="/onboarding">Cancel</a></p>''')
                return self._html(start_response, page, correlation_id, status="400 Bad Request")
            return self._html(start_response, self._page_html("Profile saved",
                '<h1>Profile saved</h1><p role="status">Your profile changes have been saved.</p><p><a href="/onboarding">Return to my account</a></p>'), 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.startswith("/api/v1/platform/tenants/") and method in {"GET", "PATCH", "POST"}:
            lifecycle = self._tenant_lifecycle_route(path, method)
            if lifecycle is not None:
                tenant, operation = lifecycle
                self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
                if self.tenant_management is None:
                    raise ValidationError("tenant management is unavailable")
                if operation == "read":
                    record = self.tenant_management.tenant(
                        tenant=tenant, correlation_id=correlation_id
                    )
                    return self._json(
                        start_response, "200 OK", _jsonable(record), correlation_id
                    )
                body = self._body(environ)
                record = self._tenant_lifecycle_change(
                    operation, tenant, body,
                    expected_version=self._expected_version(environ),
                    idempotency_key=self._idempotency_key(environ),
                    correlation_id=correlation_id,
                )
                return self._json(
                    start_response, "200 OK", _jsonable(record), 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]
            self.service.authorize_tenant_member_action(actor, user_id, tenant=tenant, correlation_id=correlation_id)
            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:
                self._change_status(actor, tenant, user_id, AccountStatus.ACTIVE,
                                    idempotency_key=idempotency_key, correlation_id=correlation_id)
                recovery = {"status": "tenant_active", "changed": ("tenant_access",)}
            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._require_tenant_admin(actor, tenant)
            self._require_invitation_tenant(invitation_id, 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.authorize_tenant_member_action(actor, user_id, tenant=tenant, correlation_id=correlation_id)
            user = self.service.store.user(user_id)
            if user is None:
                raise NotFoundError("user not found")
            self._require_setup_access(tenant, user_id)
            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")
            account = self._change_status(actor, tenant, user_id, AccountStatus.DISABLED,
                                          idempotency_key=idempotency_key, correlation_id=correlation_id)
            return self._json(start_response, "200 OK", {
                "status": "removed", "tenant_account": _jsonable(account),
                "provider_identity_removed": False,
            }, correlation_id)
        if path == "/platform/activity":
            self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
            if method != "GET":
                raise NotFoundError("activity route not found")
            self.service.tenant_diagnostics(actor, tenant=PLATFORM_TENANT, correlation_id=correlation_id)
            query = parse_qs(str(environ.get("QUERY_STRING", "")))
            return self._html(start_response, self._platform_activity(
                query.get("reference", [""])[0], query.get("tenant", [""])[0]), correlation_id)
        if path in {"/platform/operations", "/platform/operations/replay"}:
            self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
            if method == "POST" and path.endswith("/replay"):
                body = self._form_body(environ)
                self._require_csrf(environ, str(body.get("csrf_token", "")))
                event = self.service.store.outbox_event(str(body.get("event_id", "")))
                if event is None:
                    raise NotFoundError("delivery record not found")
                if event.delivered_at is not None or event.claimed_by:
                    raise ConflictError("Delivery is already completed or being processed. Refresh its status.")
                self.service.replay_outbox(actor, event.event_id, correlation_id=correlation_id)
                return self._redirect(start_response, "/platform/operations?"+urlencode({"event_id":event.event_id}), correlation_id)
            if method != "GET" or path.endswith("/replay"):
                raise NotFoundError("operations route not found")
            query = parse_qs(str(environ.get("QUERY_STRING", "")))
            return self._html(start_response, self._operations_page(actor,
                self._csrf_token(environ), query.get("event_id", [""])[0], correlation_id), 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", ""))
            with self.service.store.tenant_lifecycle_guard(tenant), self.service.store.transaction():
                existing_admin = any(
                    m.scope_type == "tenant" and m.scope_id == tenant and m.kind == "tenant-admin"
                    and (u := self.service.store.user(m.user_id)) is not None
                    and (u.primary_email or "").casefold() == email.casefold()
                    for m in self.service.store.memberships_for_tenant(tenant)
                ) if email else False
                if email and not existing_admin:
                    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 == "/platform/tenant" and method == "GET":
            self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
            query = parse_qs(str(environ.get("QUERY_STRING", "")))
            lookup = query.get("tenant", [""])[0].strip()
            view = query.get("view", ["lifecycle"])[0]
            if view not in {"lifecycle", "users"}:
                raise ValidationError("unknown tenant view")
            if not lookup.startswith("tenant:"):
                matches = tuple(
                    tenant for tenant in self._known_membership_tenants()
                    if tenant.rsplit(":", 1)[-1] == lookup
                )
                if len(matches) != 1:
                    message = (
                        "Several tenants use that name. Choose a tenant below."
                        if matches else
                        "No tenant with users matches that name. Choose a tenant below or enter its full identifier."
                    )
                    return self._html(start_response, self._platform(
                        self._csrf_token(environ), error=message,
                    ), correlation_id)
                lookup = matches[0]
            if lookup == PLATFORM_TENANT:
                raise ValidationError("a non-platform tenant identifier is required")
            return self._redirect(
                start_response,
                ("/admin/" if view == "users" else "/platform/tenants/")
                + quote(lookup, safe=""), correlation_id,
            )
        if path.startswith("/platform/tenants/") and method in {"GET", "POST"}:
            self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
            if self.tenant_management is None:
                raise ValidationError("tenant management is unavailable")
            tenant = unquote(path.split("/")[3])
            if not tenant.startswith("tenant:") or tenant == PLATFORM_TENANT:
                raise ValidationError("a non-platform tenant identifier is required")
            if method == "GET":
                record = self.tenant_management.tenant(
                    tenant=tenant, correlation_id=correlation_id
                )
                return self._html(
                    start_response,
                    self._platform_tenant(record, self._csrf_token(environ)),
                    correlation_id,
                )
            body = self._form_body(environ)
            self._require_csrf(environ, str(body.get("csrf_token", "")))
            operation = str(body.get("operation", ""))
            if operation not in {"update", "retire", "reactivate"}:
                raise ValidationError("an operation is required")
            version = str(body.get("version", ""))
            if not version.isdigit():
                raise ValidationError("the current record version is required")
            if operation in {"retire", "reactivate"}:
                current = self.tenant_management.tenant(tenant=tenant, correlation_id=correlation_id)
                preview = self._confirm_change(environ, start_response, body, str(current.version),
                    f"{operation.capitalize()} {tenant}",
                    "This changes the tenant lifecycle. Existing application sessions may take time to reflect the change. Review the tenant and reason before confirming.", correlation_id)
                if preview is not None:
                    return preview
            metadata = {
                key: str(body[key]) for key in ("display_name", "contact_email")
                if str(body.get(key, "")).strip()
            }
            record = self._tenant_lifecycle_change(
                operation, tenant, {"reason": body.get("reason"), "metadata": metadata},
                expected_version=int(version),
                # The tenant, operation, and version make the key unique per
                # logical mutation, so a resubmitted form replays rather than
                # applying the change twice.
                idempotency_key=f"portal-tenant-{operation}-{tenant}-{version}",
                correlation_id=correlation_id,
            )
            return self._html(
                start_response,
                self._platform_tenant(record, self._csrf_token(environ)),
                correlation_id,
            )
        if path.startswith("/admin/") and method == "GET":
            tenant = unquote(path.split("/")[2])
            self._require_tenant_admin(actor, tenant)
            if path.endswith("/activity"):
                self.service.tenant_diagnostics(actor, tenant=tenant, correlation_id=correlation_id)
                return self._html(start_response, self._audit_page(tenant), correlation_id)
            if len(path.split("/")) != 3:
                raise NotFoundError("tenant page not found")
            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 = unquote(parts[2])
            self._require_tenant_admin(actor, tenant)
            if len(parts) == 6 and parts[3] == "users":
                self.service.authorize_tenant_member_action(actor, parts[4], tenant=tenant, correlation_id=correlation_id)
            body = self._form_body(environ)
            self._require_csrf(environ, str(body.get("csrf_token", "")))
            if len(parts) == 6 and parts[3] == "users" and parts[5] in {"status", "remove", "recover", "role"}:
                if parts[5] == "recover":
                    self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
                user = self.service.store.user(parts[4])
                state = self.service.store.tenant_account(tenant, parts[4])
                snapshot = repr((state, self.service.store.memberships_for_user(parts[4], tenant=tenant)))
                preview = self._confirm_change(environ, start_response, body, snapshot,
                    f"{parts[5].capitalize()} account in {tenant}",
                    f"Account: {user.display_name or user.user_id}. This action applies to this tenant. Other tenant access and the shared login are retained."
                    + (" Recovery restores this tenant account and prepares a missing directory login. Verify the person's request through your established support process first. It does not reset a password, remove an authenticator, lift a global suspension, or prove account ownership. For a lost authenticator, use provider recovery; this action cannot bypass it." if parts[5] == "recover" else ""), correlation_id)
                if preview is not None:
                    return preview
            if len(parts) == 4 and parts[3] in {"users", "invitations"} and body.get("role", "user") not in {"user", "tenant-admin"}:
                raise ValidationError("Choose User or Tenant administrator.")
            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]
                self._require_invitation_tenant(invitation_id, tenant)
                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")
                self._require_setup_access(tenant, user_id)
                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, result.external_subject
                        ),
                        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] == "role":
                role = str(body.get("role", ""))
                user_id = parts[4]
                with self.service.store.tenant_lifecycle_guard(tenant):
                    self.service.validate_tenant_role_change(actor, user_id, role, tenant=tenant, correlation_id=correlation_id)
                    identity = next((i for i in self.service.store.identities_for_user(user_id) if i.provider == "netkingdom-lldap"), None)
                    account = self.service.store.tenant_account(tenant, user_id)
                    if identity is not None:
                        if not callable(getattr(self.provisioning, "tenant_access", None)):
                            raise ValidationError("Tenant-scoped identity changes are unavailable.")
                        enabled = account is not None and account.status == AccountStatus.ACTIVE
                        result = self.provisioning.tenant_access(external_subject=identity.subject, tenant=tenant,
                            roles=(role,), enabled=enabled, idempotency_key=f"portal-role-{tenant}-{user_id}-{role}", correlation_id=correlation_id)
                        if result.status != ("tenant_active" if enabled else "tenant_disabled"):
                            raise RuntimeError("tenant role change not confirmed")
                    self.service.set_tenant_role(actor, user_id, role, 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] == "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":
                self._change_status(actor, tenant, parts[4], AccountStatus.DISABLED,
                                    idempotency_key=f"portal-remove-{tenant}-{parts[4]}", 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._change_status(actor, tenant, user_id, AccountStatus.ACTIVE,
                                        idempotency_key=request.idempotency_key, correlation_id=correlation_id)
                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:
        with self.service.store.tenant_lifecycle_guard(tenant):
            return self._change_status_locked(actor, tenant, user_id, status, idempotency_key=idempotency_key, correlation_id=correlation_id)

    def _change_status_locked(
        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.authorize_tenant_member_action(actor, user_id, tenant=tenant, correlation_id=correlation_id)
        if status not in {AccountStatus.ACTIVE, AccountStatus.SUSPENDED, AccountStatus.DISABLED}:
            raise ValidationError("unsupported tenant account status")
        if status != AccountStatus.ACTIVE:
            self.service.require_admin_successor(user_id, tenant=tenant)
        identity = next((item for item in self.service.store.identities_for_user(user_id)
                         if item.provider == "netkingdom-lldap"), None)
        if identity is not None:
            if not callable(getattr(self.provisioning, "tenant_access", None)):
                raise ValidationError("Tenant-scoped identity changes are unavailable. No shared login was changed.")
            result = self.provisioning.tenant_access(
                external_subject=identity.subject, tenant=tenant,
                roles=tuple(m.kind for m in self.service.store.memberships_for_user(user_id, tenant=tenant)
                            if m.scope_type == "tenant" and m.scope_id == tenant),
                enabled=status == AccountStatus.ACTIVE,
                idempotency_key=idempotency_key, correlation_id=correlation_id,
            )
            expected = "tenant_active" if status == AccountStatus.ACTIVE else "tenant_disabled"
            if result.status != expected:
                raise RuntimeError("tenant access change not confirmed")
        return self.service.set_tenant_account_status(
            actor, user_id, status, tenant=tenant, correlation_id=correlation_id,
        )

    def _confirm_change(self, environ: Mapping[str, Any], start_response: StartResponse,
                        body: Mapping[str, str], snapshot: str, title: str, explanation: str,
                        correlation_id: str) -> list[bytes] | None:
        path = str(environ.get("PATH_INFO", ""))
        fields = {k: v for k, v in body.items() if k != "confirm_token"}
        state = hashlib.sha256(snapshot.encode()).hexdigest()
        def signature(stamp: str) -> str:
            material = json.dumps([path, fields, state, stamp], sort_keys=True).encode()
            return hmac.new(self.trusted_proxy_secret.encode(), material, hashlib.sha256).hexdigest()
        supplied = str(body.get("confirm_token", ""))
        if supplied:
            stamp, _, digest = supplied.partition(".")
            if not stamp.isdigit() or not 0 <= time.time()-int(stamp) <= 600 or not hmac.compare_digest(digest, signature(stamp)):
                raise ConflictError("The confirmation expired or the account changed. Refresh and review the action again.")
            return None
        stamp = str(int(time.time()))
        hidden = "".join(f'<input type="hidden" name="{escape(k)}" value="{escape(v)}">' for k,v in fields.items())
        details = "".join(f'<li>{escape(k.replace("_", " "))}: {escape(v)}</li>' for k,v in fields.items() if k in {"status", "operation", "reason", "version", "role"})
        page = self._page_html(title, f'<h1>{escape(title)}?</h1><p>{escape(explanation)}</p><ul>{details}</ul>'
            f'<form method="post" action="{escape(path)}">{hidden}<input type="hidden" name="confirm_token" value="{stamp}.{signature(stamp)}">'
            '<button type="submit">Confirm change</button></form><p><a href="/">Cancel without changes</a></p>')
        return self._html(start_response, page, correlation_id)

    def _audit_page(self, tenant: str) -> str:
        records = [r for r in self.service.audit_records() if r.tenant == tenant][-100:]
        rows = "".join(f'<tr><td>{escape(r.recorded_at.isoformat())}</td><td>{escape(r.action)}</td><td>{escape(r.actor.preferred_username or r.actor.subject)}</td><td>{escape(r.correlation_id)}</td></tr>' for r in reversed(records))
        return self._page_html("Account activity", f'<h1>Account activity</h1><p>Tenant: {escape(tenant)}. Most recent 100 recorded actions. A recorded request is not proof of delivery or effective application access.</p>'
            '<table><thead><tr><th>Time</th><th>Action</th><th>Actor</th><th>Support reference</th></tr></thead><tbody>'
            + (rows or '<tr><td colspan="4">No recorded activity yet.</td></tr>') + '</tbody></table>'
            f'<p><a href="/admin/{escape(tenant)}">Return to tenant administration</a></p>')

    @staticmethod
    def _delivery_status(event: Any) -> str:
        if event.delivered_at: return "Accepted by delivery adapter; receipt by the person is unverified"
        if event.dead_lettered_at: return "Delivery stopped after repeated failures"
        if event.failed_at: return "Delivery failed; retry pending"
        if event.claimed_by: return "Being processed"
        return "Queued for delivery"

    def _platform_activity(self, reference: str, tenant: str) -> str:
        reference, tenant = reference.strip(), tenant.strip()
        if len(reference) > 200 or len(tenant) > 200:
            raise ValidationError("Support reference and tenant must each be at most 200 characters.")
        entries = []
        for record in self.service.audit_records():
            if (reference and record.correlation_id != reference) or (tenant and record.tenant != tenant):
                continue
            entries.append((record.recorded_at, "Audit", record.tenant, record.action,
                record.actor.preferred_username or record.actor.subject, record.correlation_id,
                "Recorded action; external outcome is unverified", ""))
        for event in self.service.store.outbox_history():
            if (reference and event.correlation_id != reference) or (tenant and event.tenant != tenant):
                continue
            link = "/platform/operations?" + urlencode({"event_id": event.event_id})
            entries.append((event.occurred_at, "Delivery", event.tenant, event.event_type,
                "—", event.correlation_id, self._delivery_status(event), link))
        entries.sort(key=lambda row: row[0], reverse=True)
        count = len(entries)
        rows = ""
        for stamp, kind, scope, action, actor_name, ref, status, link in entries[:100]:
            detail = f'<a href="{escape(link)}">Inspect delivery</a>' if link else ""
            rows += "<tr>" + "".join(f"<td>{escape(value)}</td>" for value in
                (stamp.isoformat(), kind, scope, action, actor_name, ref, status)) + f"<td>{detail}</td></tr>"
        empty = '<tr><td colspan="8">No matching records. This does not prove that no action occurred; check the reference and provider records.</td></tr>'
        return self._page_html("Platform activity", f"""<h1>Platform activity</h1>
<p>Search an exact support reference across recorded tenant actions and delivery attempts. Add a full tenant identifier to narrow the scope.</p>
<form method="get" action="/platform/activity">
<label>Support reference <input name="reference" maxlength="200" value="{escape(reference)}"></label>
<label>Tenant identifier <input name="tenant" maxlength="200" value="{escape(tenant)}"></label>
<button type="submit">Find activity</button></form>
<p>Showing {min(count, 100)} of {count} matching records, newest first. Filters apply before the 100-record display limit.</p>
<table><thead><tr><th>Time</th><th>Kind</th><th>Tenant</th><th>Action</th><th>Actor</th><th>Support reference</th><th>Known result</th><th>Next step</th></tr></thead><tbody>{rows or empty}</tbody></table>
<p>Audit records describe recorded actions. Delivery acceptance does not prove receipt, and neither proves a provider change or rollback. Check the relevant provider before closing an incident.</p>
<p><a href="/platform/operations">Service recovery</a> · <a href="/platform">Platform administration</a></p>""")

    def _operations_page(self, actor: Any, csrf: str, event_id: str, correlation_id: str) -> str:
        self.service.tenant_diagnostics(actor, tenant=PLATFORM_TENANT, correlation_id=correlation_id)
        events = list(self.service.store.outbox_history())[-100:]
        if event_id:
            event = self.service.store.outbox_event(event_id)
            if event is None: raise NotFoundError("delivery record not found")
            events = [event]
        rows = ""
        for event in events:
            action = ""
            if event.delivered_at is None and not event.claimed_by and (event.failed_at or event.dead_lettered_at):
                action = f'<form method="post" action="/platform/operations/replay"><input type="hidden" name="csrf_token" value="{escape(csrf)}"><input type="hidden" name="event_id" value="{escape(event.event_id)}"><button type="submit">Queue a retry</button></form>'
            rows += f'<tr><td>{escape(event.event_id)}</td><td>{escape(event.tenant)}</td><td>{escape(event.event_type)}</td><td>{escape(self._delivery_status(event))}</td><td>{escape(event.correlation_id)}</td><td>{action}</td></tr>'
        configuration = self._operation_capabilities()
        return self._page_html("Service recovery", '<h1>Service recovery</h1>' + configuration + '<p><a href="/platform/activity">Investigate a support reference</a></p><p>This view shows local delivery records. Live sign-in, email receipt and authenticator health are not verified here.</p>'
            '<form method="get" action="/platform/operations"><label>Delivery record ID <input name="event_id"></label><button type="submit">Find delivery</button></form>'
            '<table><thead><tr><th>Delivery</th><th>Tenant</th><th>Kind</th><th>Status</th><th>Support reference</th><th>Recovery</th></tr></thead><tbody>'
            + (rows or '<tr><td colspan="6">No delivery records. This does not prove mail was received.</td></tr>')
            + '</tbody></table><p>Queued retries are processed by the delivery worker. Check the record again for the result.</p><p><a href="/platform">Return to platform administration</a></p>')

    def _operation_capabilities(self) -> str:
        capabilities = (
            ("Portal sign-in", self.oidc_client is not None, "An existing portal session does not prove a fresh provider sign-in works."),
            ("Tenant identity management", self.provisioning is not None, "Use tenant administration for login setup or tenant access recovery. Shared identity and factor recovery belong to the sign-in service."),
            ("Tenant lifecycle", self.tenant_management is not None, "Review the authority's returned version after a change."),
            ("Notification delivery", self.outbox_delivery is not None, "Inspect the delivery record below. If email cannot be received, use the tenant's assisted password setup process."),
        )
        rows = "".join(f'<tr><td>{escape(name)}</td><td>{"Configured; live health unverified" if configured else "Unavailable in this portal"}</td><td>{escape(help_text)}</td></tr>'
                       for name, configured, help_text in capabilities)
        return ('<section><h2>Service capabilities</h2><table><thead><tr><th>Service</th><th>Known state</th><th>Recovery step</th></tr></thead><tbody>'
            + rows + '</tbody></table><p>Authenticator recovery and authentication policy changes are unavailable in this portal. The sign-in service owner must verify factor lookup, recovery and policy enforcement. A configured adapter is not a health check.</p></section>')

    def _require_setup_access(self, tenant: str, user_id: str) -> None:
        account = self.service.store.tenant_account(tenant, user_id)
        if account is not None and account.status in {AccountStatus.SUSPENDED, AccountStatus.DISABLED}:
            raise ConflictError("Reactivate this tenant account before creating a password setup link.")

    def _require_tenant_admin(self, actor: Any, tenant: str) -> None:
        self.service.resolve_tenant_context(actor, tenant)
        if not {"tenant-admin", "platform-operator"}.intersection(actor.roles):
            raise AuthorizationDenied("tenant administrator role required")

    def _require_invitation_tenant(self, invitation_id: str, tenant: str) -> None:
        invitation = self.service.store.family_invitation(invitation_id)
        if invitation is None or invitation.tenant != tenant:
            raise NotFoundError("invitation not found in this tenant")

    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")

        raw_idempotency_key = str(
            body.get("idempotency_key") if browser
            else environ.get("HTTP_IDEMPOTENCY_KEY", "")
        )
        if len(raw_idempotency_key) < 16 or len(raw_idempotency_key) > 256:
            raise ValidationError("registration idempotency key is invalid")
        idempotency_hash = hmac.new(
            self.trusted_proxy_secret.encode(), raw_idempotency_key.encode(),
            hashlib.sha256,
        ).hexdigest()
        request_hash = hashlib.sha256(json.dumps({
            "username": username, "email": email, "display_name": display_name,
            "client_id": client_id, "tenant": tenant,
        }, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
        previous = next((
            item for item in self.service.store.all_registration_sessions()
            if item.start_idempotency_hash
            and hmac.compare_digest(item.start_idempotency_hash, idempotency_hash)
        ), None)
        if previous is not None:
            if not previous.start_request_hash or not hmac.compare_digest(
                previous.start_request_hash, request_hash
            ):
                raise ConflictError("registration idempotency key was reused")
            return self._registration_requested_response(
                start_response, correlation_id, browser
            )

        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,
        )
        session = replace(
            session, start_idempotency_hash=idempotency_hash,
            start_request_hash=request_hash,
        )
        self.service.store.save_registration_session(session)
        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,
            )
        )
        return self._registration_requested_response(
            start_response, correlation_id, browser
        )

    def _registration_requested_response(
        self, start_response: StartResponse, correlation_id: str, browser: bool
    ) -> Iterable[bytes]:
        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 _cancel_public_registration(
        self, environ, start_response, correlation_id, *,
        body: Mapping[str, Any] | None = None, browser: bool = False,
    ):
        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.cancel(
            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("cancellation 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,
        )
        self.service.abandon_registration(
            actor, session.registration_id, correlation_id=correlation_id
        )
        if browser:
            return self._html(
                start_response,
                self._page_html(
                    "Registration canceled",
                    "<h1>Registration canceled.</h1>"
                    "<p>The request can no longer be verified. You may start again at any time.</p>"
                    '<p><a class="button" href="/register">Start again</a></p>',
                ),
                correlation_id,
            )
        return self._json(
            start_response, "200 OK", {"status": "registration_canceled"},
            correlation_id,
        )

    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 _tenant_lifecycle_route(path: str, method: str) -> tuple[str, str] | None:
        """Match the authority-backed lifecycle routes, not the recovery route."""
        parts = path.split("/")[5:]
        if not parts or not parts[0]:
            return None
        tenant = unquote(parts[0])
        if not tenant.startswith("tenant:") or tenant == PLATFORM_TENANT:
            return None
        if len(parts) == 1:
            if method == "GET":
                return tenant, "read"
            if method == "PATCH":
                return tenant, "update"
            return None
        if len(parts) == 2 and method == "POST" and parts[1] in {"retire", "reactivate"}:
            return tenant, parts[1]
        return None

    def _tenant_lifecycle_change(
        self, operation: str, tenant: str, body: Mapping[str, Any], *,
        expected_version: int, idempotency_key: str, correlation_id: str,
    ) -> Any:
        reason = str(body.get("reason") or "").strip()
        if not reason:
            raise ValidationError("a reason is required for a tenant lifecycle change")
        assert self.tenant_management is not None
        if operation == "update":
            metadata = body.get("metadata")
            if not isinstance(metadata, Mapping):
                raise ValidationError("metadata must be an object")
            return self.tenant_management.update_tenant(
                tenant=tenant,
                metadata={str(key): str(value) for key, value in metadata.items()},
                expected_version=expected_version, reason=reason,
                idempotency_key=idempotency_key, correlation_id=correlation_id,
            )
        change = (
            self.tenant_management.retire_tenant
            if operation == "retire"
            else self.tenant_management.reactivate_tenant
        )
        return change(
            tenant=tenant, expected_version=expected_version, reason=reason,
            idempotency_key=idempotency_key, correlation_id=correlation_id,
        )

    @staticmethod
    def _expected_version(environ: Mapping[str, Any]) -> int:
        value = str(environ.get("HTTP_IF_MATCH", "")).strip().strip('"')
        if not value.isdigit():
            raise ValidationError("an If-Match record 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 _identity_switch_help(self) -> str:
        if not self.oidc_client:
            return ""
        return (
            '<details><summary>Wrong account appears when signing in?</summary>'
            '<p>You can clear the shared sign-in before choosing another account. '
            'Other applications may keep their own sessions.</p>'
            f'<p><a href="{escape(self.oidc_client.issuer)}/account/logout">'
            'Use another account</a></p></details>'
        )

    def _security_page(self) -> str:
        handoff = (
            '<p><a class="button" rel="noreferrer" href="'
            + escape(self.mfa_management_url)
            + '">Manage authenticator app</a></p>'
            '<p>The sign-in service will ask you to verify your identity. Check the account name there before making changes.</p>'
            if self.mfa_management_url else
            '<p role="status">Authenticator setup is temporarily unavailable. '
            'If a code is requested before you have set up an authenticator, contact your tenant administrator.</p>'
        )
        return self._page_html("Sign-in security", """
<h1>Sign-in security</h1>
<p>Use this page for help with your password and authenticator app.</p>
<section><h2>Two-step verification</h2>
<p>An authenticator app generates a short-lived code to enter after your password.
This portal cannot currently confirm whether an authenticator is enabled for your account.</p>
""" + handoff + """
<details><summary>How to set up an authenticator when setup is available</summary>
<ol><li>Open authenticator management and check that it shows your account.</li>
<li>Choose to add an authenticator and scan its QR code with your authenticator app.</li>
<li>Enter a current code to confirm setup. Wait for the sign-in service to confirm activation.</li>
<li>Follow the recovery instructions shown there, then test a new sign-in before closing your current session.</li></ol>
<p>Opening the setup page does not activate two-step verification. If you cancel, check the status in authenticator management before leaving.</p></details>
<details><summary>A code is rejected, or I have lost my authenticator</summary>
<p>Use the newest code for the correct account and check that your device sets its time automatically.
If you cannot use your authenticator, follow the sign-in service's recovery instructions or ask your tenant administrator for account recovery.
Never send your password, QR code, or verification codes to an administrator.</p></details></section>
<section><h2>Password help</h2><p>Use password recovery on the sign-in page.
If the recovery message does not arrive, ask your tenant administrator for a new password setup link.
Use the login name they provide; it may differ from your display name.</p></section>
<p><a href="/access-recovery">Back to sign-in help</a></p>""")

    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">View my account</a></p>'
            if actor is not None
            else (
                '<p>You are not signed in to this portal.</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, idempotency_key: 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)}">
<input type="hidden" name="idempotency_key" value="{escape(idempotency_key)}">
<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 _registration_cancel_form(self, csrf_token: str, handle: str) -> str:
        return self._page_html(
            "Cancel registration",
            f"""<h1>Cancel this registration?</h1>
<p>This verification request will stop working. No login identity will be created.</p>
<form method="post" action="/registration/cancel">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<input type="hidden" name="handle" value="{escape(handle)}">
<button type="submit">Cancel registration</button></form>
<p><a href="/register">Keep registration</a></p>""",
        )
    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 if item.scope_type == "tenant" and item.scope_id == tenant
        ) 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>'
        progress_items = []
        for journey in self.service.store.onboarding_journeys_for_tenant(tenant):
            user = self.service.store.user(journey.user_id)
            name = user.display_name or user.user_id if user else "Account"
            pending = [step for step in journey.steps if step.status.value not in {"completed", "skipped", "cancelled"}]
            steps = "; ".join(step.title + " — " + step.status.value.replace("_", " ") for step in pending)
            progress_items.append(f'<li>{escape(name)}: {escape(journey.status.value.replace("_", " "))}. {escape(steps)}'
                                  '<p>Ask the person to open My account for profile steps, or use sign-in help for provider steps.</p></li>')
        progress = "".join(progress_items) or "<li>No additional onboarding journeys are recorded.</li>"
        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>
<p><a href="/admin/{escape(tenant)}/activity">Account activity and support references</a> · <a href="/security">Sign-in recovery help</a></p>
<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>Check the account and invitation states below. Password and authenticator status are not available here.</p><ul>{diagnostic_items}</ul></section>
<section><h2>Onboarding follow-up</h2><ul>{progress}</ul><p>These are recorded workflow states. Missing password or authenticator evidence is not proof of completion.</p></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 "—"
        events = [e for e in self.service.store.outbox_history() if e.tenant == tenant
                  and e.aggregate_id == invitation.invitation_id and e.event_type in {"family_invitation.created", "family_invitation.resent", "family_member.invited"}]
        delivery = self._delivery_status(events[-1]) if events else "Delivery status unavailable"
        return (
            f"<tr><td>{escape(invitation.primary_email)}</td><td>{escape(invitation.role)}</td>"
            f"<td>{escape(invitation.status.value)}<p>{escape(delivery)}</p></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)
        directory = next((i for i in self.service.store.identities_for_user(membership.user_id)
                          if i.provider == "netkingdom-lldap"), None)
        account = self.service.store.tenant_account(tenant, membership.user_id)
        status = account.status if account else AccountStatus.INVITED
        inactive = status in {AccountStatus.SUSPENDED, AccountStatus.DISABLED}
        root = f"/admin/{quote(tenant, safe='')}/users/{quote(membership.user_id, safe='')}"
        def form(action: str, label: str, **fields: str) -> str:
            hidden = "".join(f'<input type="hidden" name="{escape(k)}" value="{escape(v)}">' for k,v in fields.items())
            return f'<form method="post" action="{root}/{action}"><input type="hidden" name="csrf_token" value="{escape(csrf_token)}">{hidden}<button type="submit">{label}</button></form>'
        login = (f'<p>Login name: <strong>{escape(self._directory_login(directory.subject))}</strong></p>'
                 '<p>Password and authenticator status are not available here.</p>') if directory else '<p>Login not created. Prepare a login before asking this person to sign in.</p>'
        actions = ""
        if not inactive:
            actions += form("provision", "Create password setup link" if directory else "Create login")
        actions += form("status", "Reactivate" if inactive else "Suspend", status="active" if inactive else "suspended")
        if status != AccountStatus.DISABLED:
            actions += form("remove", "Remove account")
        next_role = "user" if membership.kind == "tenant-admin" else "tenant-admin"
        actions += form("role", "Make user" if next_role == "user" else "Make tenant administrator", role=next_role)
        if platform_operator: actions += form("recover", "Restore tenant account")
        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><td>{escape(membership.kind)}</td>'
                f'<td>{escape(status.value)} for this tenant</td><td>{login}</td><td>{actions}</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 _known_membership_tenants(self) -> tuple[str, ...]:
        # A navigation index of user-owned memberships, not an authority inventory.
        return tuple(
            tenant for tenant in self.service.store.membership_tenants()
            if tenant.startswith("tenant:") and tenant != PLATFORM_TENANT
        )

    def _platform(self, csrf_token: str, *, error: str = "") -> str:
        known_tenants = "".join(
            f'<li><a href="/admin/{escape(quote(tenant, safe=""))}">'
            f'{escape(tenant.rsplit(":", 1)[-1])}</a> '
            f'<small>({escape(tenant)})</small></li>'
            for tenant in self._known_membership_tenants()
        ) or '<li>No tenants with user memberships are recorded yet.</li>'
        message = f'<p role="alert">{escape(error)}</p>' if error else ""
        return self._page_html(
            "Platform administration",
            f"""<h1>Platform administration</h1>
<section aria-labelledby="manage-tenant"><h2 id="manage-tenant">Manage an existing tenant</h2>
{message}
<p>Choose a tenant to manage its users:</p><ul>{known_tenants}</ul>
<form method="get" action="/platform/tenant">
<label>Tenant name or full identifier <input name="tenant" required placeholder="demo-company" aria-describedby="tenant-lookup-help"></label>
<p id="tenant-lookup-help">Enter a listed name, such as demo-company. You can also use a full identifier, such as tenant:trial:demo-company.</p>
<button type="submit" name="view" value="users">Manage users</button>
<button type="submit" name="view" value="lifecycle">Open tenant lifecycle</button></form></section>
<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_tenant(self, record: Any, csrf_token: str) -> str:
        retired = record.lifecycle == "retired"
        transition = "reactivate" if retired else "retire"
        replayed = (
            "<p>This result was replayed from the original mutation; nothing changed twice.</p>"
            if record.replayed else ""
        )
        hidden = (
            f'<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">'
            f'<input type="hidden" name="version" value="{record.version}">'
        )
        metadata_form = "" if retired else f"""<section aria-labelledby="tenant-metadata"><h2 id="tenant-metadata">Metadata</h2>
<form method="post" action="/platform/tenants/{escape(quote(record.tenant, safe=''))}">{hidden}
<input type="hidden" name="operation" value="update">
<label>Display name <input name="display_name" value="{escape(record.display_name or '')}"></label>
<label>Contact email <input name="contact_email" type="email" value="{escape(record.contact_email or '')}"></label>
<label>Reason <input name="reason" required></label>
<button type="submit">Save metadata</button></form>
<p>Only the display name and contact email are mutable; the identifier is minted into tokens.</p></section>"""
        return self._page_html(
            f"Tenant {record.tenant}",
            f"""<h1>{escape(record.tenant)}</h1>
<p>Lifecycle <strong>{escape(record.lifecycle)}</strong> at version {record.version}.</p>
<p><a class="button" href="/admin/{escape(quote(record.tenant, safe=''))}">Manage users</a></p>
<p>Grouping <strong>{escape(record.grouping or 'not reported')}</strong>, as reported by the tenant authority. The identifier's own segment is historical after a reclassification and is not the grouping.</p>
{replayed}
{metadata_form}
<section aria-labelledby="tenant-lifecycle"><h2 id="tenant-lifecycle">Lifecycle</h2>
<form method="post" action="/platform/tenants/{escape(quote(record.tenant, safe=''))}">{hidden}
<input type="hidden" name="operation" value="{transition}">
<label>Reason <input name="reason" required></label>
<button type="submit">{'Reactivate tenant' if retired else 'Retire tenant'}</button></form>
<p>Retirement is reversible and preserves grant and plan history; there is no hard delete.</p></section>
<p><a href="/platform">Return to platform administration</a></p>""",
        )

    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:
        platform_operator = "platform-operator" in session.actor.roles
        empty_memberships = (
            "<li>You have no personal tenant memberships. Your platform operator role lets you manage tenants through platform administration.</li>"
            if platform_operator else "<li>No tenant memberships yet.</li>"
        )
        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 empty_memberships
        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">Current identity</h2><p>Signed in as <strong>{escape(session.actor.preferred_username or session.actor.subject)}</strong>.</p><p>Sign-in tenant: {escape(session.actor.tenant)}.</p><p>Roles: {escape(", ".join(session.actor.roles) or "None")}.</p><p>{escape(verification)}</p><p><a href="/security">Password and two-step verification help</a></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="workloads"><h2 id="workloads">Workload access</h2>{self._workload_memberships(memberships)}<p>Each application checks access when you open it. Tenant membership alone does not grant access to every application.</p></section>
<section aria-labelledby="steps"><h2 id="steps">Onboarding progress</h2><ul>{journey_items}</ul></section>""",
        )

    @staticmethod
    def _workload_memberships(memberships: tuple[Any, ...]) -> str:
        items = "".join(
            f"<li>{escape(item.scope_id)} — {escape(item.kind)} ({escape(item.tenant)})</li>"
            for item in memberships if item.scope_type in {"application", "service", "workload", "asset"}
        )
        return "<ul>" + items + "</ul>" if items else "<p>No workload-specific access is recorded for this account.</p>"

    @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>"
        )

    @staticmethod
    def _directory_login(subject: str) -> str:
        match = re.fullmatch(r"uid=([A-Za-z0-9._-]+),ou=people,dc=netkingdom,dc=local", subject)
        return match.group(1) if match else subject

    def _password_setup_handoff(self, setup_url: str, tenant: str, subject: str = "") -> str:
        if not setup_url.startswith("https://"):
            raise ValidationError("password setup handoff must use HTTPS")
        return self._page_html(
            "Password setup",
            "<h1>Login ready for password setup</h1>"
            + f"<p>Login name: <strong>{escape(self._directory_login(subject))}</strong>. Use this name when signing in; it may differ from the display name.</p>"
            + "<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""]

    def _set_account_navigation(self, environ: Mapping[str, Any], actor: Any | None) -> None:
        if actor is None:
            _ACCOUNT_NAVIGATION.set('<p>Not signed in to this portal</p><nav aria-label="Account navigation"><a href="/">Home</a><a href="/security">Sign-in help</a><a class="button" href="/login">Sign in</a></nav>')
            return
        links = '<a href="/">Home</a><a href="/onboarding">My account</a><a href="/security">Sign-in security</a>'
        if "platform-operator" in actor.roles:
            links += '<a href="/platform">Platform administration</a><a href="/platform/operations">Service recovery</a><a href="/platform/activity">Platform activity</a>'
        elif "tenant-admin" in actor.roles:
            links += f'<a href="/admin/{escape(quote(actor.tenant, safe=""))}">Manage users</a>'
        session_id = cookie_value(str(environ.get("HTTP_COOKIE", "")), "ue_session")
        csrf = self.oidc_client.csrf_token(session_id or "") if self.oidc_client else None
        if csrf:
            links += '<a href="/logout">Log out</a>'
        identity = escape(actor.preferred_username or actor.subject)
        _ACCOUNT_NAVIGATION.set(f'<p>Signed in to this portal as <strong>{identity}</strong></p><nav aria-label="Account navigation">' + links + '</nav>')

    @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)}}
header nav{{display:flex;align-items:center;flex-wrap:wrap;gap:.75rem 1.25rem;margin-top:.75rem}}header nav form{{margin:0}}header nav button{{padding:.45rem .8rem}}
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>{_ACCOUNT_NAVIGATION.get()}</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, status: str = "200 OK",
    ) -> list[bytes]:
        data = body.encode()
        start_response(status, [
            ("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]:
        if _BROWSER_REQUEST.get():
            title = "This action could not be completed"
            guidance = "Check your account and access, then try again."
            if code == "access_denied":
                title = "Access is not available"
                guidance = "You may need to sign in again, or your account may not have permission for this action."
            elif code == "provisioning_unavailable":
                guidance = "Account services are temporarily unavailable. Check the account status before retrying."
            page = self._page_html(title, f'<h1>{title}</h1><p role="alert">{guidance}</p>'
                '<p><a href="/access-recovery">Account and sign-in help</a></p>'
                f'<p>If you need help, give your administrator this reference: <code>{escape(str(correlation_id))}</code>.</p>')
            data = page.encode()
            start_response(status, [("Content-Type", "text/html; charset=utf-8"), ("Content-Length", str(len(data))), *self._security_headers(correlation_id)])
            return [data]
        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'"),
        ]