user-engine/tests/test_web.py

import io
import hashlib
import json
import re
import unittest
from dataclasses import replace
from datetime import timedelta
from urllib.parse import quote, urlencode
from urllib.error import HTTPError
from unittest.mock import patch

from user_engine.adapters.tenant_management import HTTPTenantManagementAdapter

from user_engine.adapters import InMemoryUserEngineStore, LocalAuthorizationCheckPort
from user_engine.domain import (
    OnboardingJourney, OnboardingJourneyStatus, OnboardingStep,
    OnboardingStepStatus, OnboardingTriggerType, SubsystemHandoff,
)
from user_engine.oidc import BrowserSession, OIDCClient
from user_engine.ports import (
    IdentityDriftResult,
    ProvisioningResult,
    RegistrationVerificationReceipt,
    TenantProvisioningResult,
    TenantRecord,
    VerifiedRegistrationApplicant,
)
from user_engine.errors import ConflictError, NotFoundError, ValidationError
from user_engine.service import UserEngineService
from user_engine.testing.fixtures import FixtureIdentityClaimsAdapter, human_actor_claims
from user_engine.web import PortalApplication
from user_engine.domain import utc_now


SECRET = "test-proxy-secret-with-adequate-length"


def invoke(
    app, path, *, method="GET", claims=None, marker=SECRET, body=None,
    form=None, cookie=None, headers=None, query="", remote_addr="127.0.0.1",
):
    payload = (
        urlencode(form).encode()
        if form is not None
        else json.dumps(body or {}).encode()
    )
    environ = {
        "REQUEST_METHOD": method,
        "PATH_INFO": path,
        "QUERY_STRING": query,
        "CONTENT_LENGTH": str(len(payload)),
        "wsgi.input": io.BytesIO(payload),
        "HTTP_X_REQUEST_ID": "corr_test",
        "REMOTE_ADDR": remote_addr,
    }
    if form is not None:
        environ["CONTENT_TYPE"] = "application/x-www-form-urlencoded"
    if cookie is not None:
        environ["HTTP_COOKIE"] = cookie
    if claims is not None:
        environ["HTTP_X_VERIFIED_OIDC_CLAIMS"] = json.dumps(claims)
        environ["HTTP_X_USER_ENGINE_PROXY_SECRET"] = marker
    if path == "/api/v1/public/registrations" and method == "POST":
        environ["HTTP_IDEMPOTENCY_KEY"] = hashlib.sha256(payload).hexdigest()
    environ.update(headers or {})
    captured = {}

    def start_response(status, headers):
        captured["status"] = status
        captured["headers"] = dict(headers)

    response = b"".join(app(environ, start_response))
    return captured, response


class PortalApplicationTests(unittest.TestCase):
    def setUp(self):
        store = InMemoryUserEngineStore()
        store.migrate()
        service = UserEngineService(
            store=store,
            identity_adapter=FixtureIdentityClaimsAdapter(),
            authorization=LocalAuthorizationCheckPort(),
        )
        self.app = PortalApplication(
            service,
            trusted_proxy_secret=SECRET,
            login_url="https://kc.example/login",
        )
        self.claims = human_actor_claims(tenant="tenant:friendly:binky")

    def platform_claims(self):
        claims = human_actor_claims(subject="platform-operator", tenant="platform:root")
        claims["roles"] = ["platform-operator"]
        return claims

    def test_public_health_and_home(self):
        health, payload = invoke(self.app, "/healthz")
        self.assertEqual("200 OK", health["status"])
        self.assertEqual("no-store", health["headers"]["Cache-Control"])
        self.assertEqual("ok", json.loads(payload)["status"])
        home, html = invoke(self.app, "/")
        self.assertEqual("200 OK", home["status"])
        self.assertIn(b"Sign in with KeyCape", html)
        self.assertIn(b'name="viewport"', html)
        self.assertIn(b"focus-visible", html)
        self.assertIn(b"<main>", html)

    def test_metrics_expose_only_bounded_aggregate_state(self):
        denied, _ = invoke(self.app, "/metrics", claims={}, marker="")
        self.assertEqual("403 Forbidden", denied["status"])
        result, payload = invoke(self.app, "/metrics", claims={})
        self.assertEqual("200 OK", result["status"])
        self.assertEqual(
            "text/plain; version=0.0.4; charset=utf-8",
            result["headers"]["Content-Type"],
        )
        self.assertIn(b"user_engine_ready 1", payload)
        self.assertIn(b'user_engine_records{kind="users"} 0', payload)
        self.assertNotIn(SECRET.encode(), payload)

    def test_protected_route_rejects_untrusted_claim_header(self):
        result, payload = invoke(
            self.app, "/api/v1/me", claims=self.claims, marker="attacker"
        )
        self.assertEqual("403 Forbidden", result["status"])
        self.assertNotIn(b"attacker", payload)

    def test_expired_browser_session_and_provider_outage_fail_closed(self):
        oidc = OIDCClient(
            issuer="https://kc.example", client_id="portal",
            redirect_uri="https://users.example/oidc/callback", audience="portal",
        )
        oidc.sessions["expired"] = BrowserSession(
            claims=self.claims, expires_at=0, csrf_token="expired-csrf"
        )
        self.app.oidc_client = oidc
        expired, _ = invoke(
            self.app, "/onboarding", cookie="ue_session=expired"
        )
        self.assertEqual("403 Forbidden", expired["status"])

        self.app.oidc_client = None
        self.app.provisioning = FailingProvisioning()
        created, payload = invoke(
            self.app, "/api/v1/tenants/tenant:friendly:binky/users",
            method="POST", claims=self.claims,
            body={"primary_email": "failure@example.test", "role": "user"},
        )
        user_id = json.loads(payload)["user"]["user_id"]
        before = self.app.service.store.tenant_account(
            "tenant:friendly:binky", user_id
        ).status
        failed, payload = invoke_with_idempotency(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/users/{user_id}/provision",
            self.claims,
        )
        self.assertEqual("502 Bad Gateway", failed["status"])
        self.assertNotIn(b"provider-secret", payload)
        self.assertEqual(
            before,
            self.app.service.store.tenant_account(
                "tenant:friendly:binky", user_id
            ).status,
        )

    def test_verified_claims_create_current_user(self):
        result, payload = invoke(self.app, "/api/v1/me", claims=self.claims)
        self.assertEqual("200 OK", result["status"])
        decoded = json.loads(payload)
        self.assertEqual("tenant:friendly:binky", decoded["actor"]["tenant"])

        updated, payload = invoke_with_idempotency(
            self.app, "/api/v1/me/profile", self.claims, method="PATCH",
            body={
                "display_name": "Sample Person", "consent_accepted": True,
                "consent_version": "portal-terms-v1",
            },
        )
        self.assertEqual("200 OK", updated["status"])
        self.assertEqual("Sample Person", json.loads(payload)["display_name"])

    def test_registration_api_is_correlated(self):
        result, payload = invoke(
            self.app,
            "/api/v1/registrations",
            method="POST",
            claims=self.claims,
            body={"tenant": "tenant:friendly:binky"},
        )
        self.assertEqual("201 Created", result["status"])
        self.assertEqual("corr_test", result["headers"]["X-Request-ID"])
        self.assertEqual("factor_pending", json.loads(payload)["status"])

    def test_public_registration_start_and_verified_factor_are_bound(self):
        verifier = FakeRegistrationVerification()
        self.app.registration_verification = verifier
        self.app.provisioning = FakeProvisioning()
        self.app.registration_clients = frozenset({"coulomb-social"})
        self.app.registration_tenants = frozenset({"tenant:coulomb"})
        self.app.registration_oidc_issuer = "https://kc.example"
        self.app.registration_password_setup_origins = frozenset(
            {"https://kc.example"}
        )
        started, payload = invoke(
            self.app,
            "/api/v1/public/registrations",
            method="POST",
            body={
                "username": "New.Person",
                "email": "New.Person@Example.Test",
                "display_name": "New Person",
                "client_id": "coulomb-social",
                "tenant": "tenant:coulomb",
                "return_to": "https://evil.example",
            },
        )
        self.assertEqual("202 Accepted", started["status"])
        self.assertEqual({"status": "verification_requested"}, json.loads(payload))
        self.assertEqual("new.person", verifier.requested.preferred_username)
        self.assertEqual("new.person@example.test", verifier.requested.normalized_email)

        verifier.registration_id = verifier.requested.registration_id
        verified, payload = invoke(
            self.app,
            "/api/v1/public/registrations/verify",
            method="POST",
            body={"handle": "x" * 32},
        )
        self.assertEqual("303 See Other", verified["status"])
        self.assertEqual(
            "https://kc.example/setup/password?token=opaque",
            verified["headers"]["Location"],
        )
        self.assertEqual(1, self.app.service.operability_snapshot().metrics["users"])
        session = self.app.service.store.registration_session(verifier.registration_id)
        self.assertEqual("completed", session.status.value)
        identities = self.app.service.store.identities_for_user(session.user_id)
        self.assertTrue(any(
            identity.issuer == "https://kc.example"
            and identity.subject == "new.person"
            for identity in identities
        ))
        provision = self.app.provisioning.requests[-1]
        self.assertEqual(("user",), provision.roles)
        self.assertEqual("new.person", provision.preferred_username)
        self.assertEqual(
            f"public-registration-{verifier.registration_id}",
            provision.idempotency_key,
        )

    def test_public_registration_browser_journey_uses_csrf_and_confirmation(self):
        verifier = FakeRegistrationVerification()
        self.app.registration_verification = verifier
        self.app.provisioning = FakeProvisioning()
        self.app.registration_clients = frozenset({"coulomb-social"})
        self.app.registration_tenants = frozenset({"tenant:coulomb"})
        self.app.registration_oidc_issuer = "https://kc.example"
        self.app.registration_password_setup_origins = frozenset({"https://kc.example"})

        page, html = invoke(self.app, "/register")
        self.assertEqual("200 OK", page["status"])
        self.assertIn(b"Create your account", html)
        cookie = page["headers"]["Set-Cookie"]
        token = re.search(rb'name="csrf_token" value="([^"]+)"', html).group(1).decode()
        idempotency_key = re.search(
            rb'name="idempotency_key" value="([^"]+)"', html
        ).group(1).decode()
        denied, _ = invoke(self.app, "/register", method="POST", form={
            "csrf_token": "wrong", "username": "new.person",
            "email": "new@example.test", "client_id": "coulomb-social",
            "tenant": "tenant:coulomb",
        }, cookie=cookie)
        self.assertEqual("403 Forbidden", denied["status"])
        started, html = invoke(self.app, "/register", method="POST", form={
            "csrf_token": token, "idempotency_key": idempotency_key,
            "username": "new.person",
            "email": "new@example.test", "display_name": "New Person",
            "client_id": "coulomb-social", "tenant": "tenant:coulomb",
        }, cookie=cookie)
        self.assertEqual("200 OK", started["status"])
        self.assertIn(b"Check your email", html)

        verifier.registration_id = verifier.requested.registration_id
        confirmation, html = invoke(
            self.app, "/registration/verify", query="handle=" + "x" * 32
        )
        self.assertEqual("200 OK", confirmation["status"])
        self.assertIn(b"Verify and continue", html)
        cookie = confirmation["headers"]["Set-Cookie"]
        token = re.search(rb'name="csrf_token" value="([^"]+)"', html).group(1).decode()
        completed, html = invoke(
            self.app, "/registration/verify", method="POST", cookie=cookie,
            form={"csrf_token": token, "handle": "x" * 32},
        )
        self.assertEqual("200 OK", completed["status"])
        self.assertIn(b"Create password", html)

    def test_public_registration_browser_cancellation_is_bound_and_terminal(self):
        verifier = FakeRegistrationVerification()
        self.app.registration_verification = verifier
        self.app.registration_clients = frozenset({"coulomb-social"})
        self.app.registration_tenants = frozenset({"tenant:coulomb"})
        invoke(self.app, "/api/v1/public/registrations", method="POST", body={
            "username": "cancel.person", "email": "cancel@example.test",
            "client_id": "coulomb-social", "tenant": "tenant:coulomb",
        })
        verifier.registration_id = verifier.requested.registration_id
        confirmation, html = invoke(
            self.app, "/registration/cancel", query="handle=" + "x" * 32
        )
        cookie = confirmation["headers"]["Set-Cookie"]
        token = re.search(rb'name="csrf_token" value="([^"]+)"', html).group(1).decode()
        canceled, html = invoke(
            self.app, "/registration/cancel", method="POST", cookie=cookie,
            form={"csrf_token": token, "handle": "x" * 32},
        )
        self.assertEqual("200 OK", canceled["status"])
        self.assertIn(b"Registration canceled", html)
        session = self.app.service.store.registration_session(verifier.registration_id)
        self.assertEqual("abandoned", session.status.value)
        verify, _ = invoke(
            self.app, "/api/v1/public/registrations/verify", method="POST",
            body={"handle": "x" * 32}, remote_addr="127.0.0.2",
        )
        self.assertEqual("400 Bad Request", verify["status"])

    def test_public_registration_rejects_untrusted_password_setup_origin(self):
        verifier = FakeRegistrationVerification()
        self.app.registration_verification = verifier
        self.app.provisioning = FakeProvisioning(
            password_setup_url="https://evil.example/setup"
        )
        self.app.registration_clients = frozenset({"coulomb-social"})
        self.app.registration_tenants = frozenset({"tenant:coulomb"})
        self.app.registration_oidc_issuer = "https://kc.example"
        self.app.registration_password_setup_origins = frozenset(
            {"https://kc.example"}
        )
        invoke(
            self.app,
            "/api/v1/public/registrations",
            method="POST",
            body={
                "username": "person",
                "email": "person@example.test",
                "client_id": "coulomb-social",
                "tenant": "tenant:coulomb",
            },
        )
        verifier.registration_id = verifier.requested.registration_id
        result, _ = invoke(
            self.app,
            "/api/v1/public/registrations/verify",
            method="POST",
            body={"handle": "x" * 32},
        )
        self.assertEqual("400 Bad Request", result["status"])

    def test_public_registration_resumes_after_provider_outage(self):
        verifier = FakeRegistrationVerification()
        provisioning = FailingOnceProvisioning()
        self.app.registration_verification = verifier
        self.app.provisioning = provisioning
        self.app.registration_clients = frozenset({"coulomb-social"})
        self.app.registration_tenants = frozenset({"tenant:coulomb"})
        self.app.registration_oidc_issuer = "https://kc.example"
        self.app.registration_password_setup_origins = frozenset({"https://kc.example"})
        invoke(self.app, "/api/v1/public/registrations", method="POST", body={
            "username": "retry.person", "email": "retry@example.test",
            "client_id": "coulomb-social", "tenant": "tenant:coulomb",
        })
        verifier.registration_id = verifier.requested.registration_id
        pending, payload = invoke(
            self.app, "/api/v1/public/registrations/verify", method="POST",
            body={"handle": "x" * 32},
        )
        self.assertEqual("202 Accepted", pending["status"])
        recovery = json.loads(payload)
        resumed, _ = invoke(
            self.app, "/api/v1/public/registrations/resume", method="POST",
            body={"registration_id": recovery["registration_id"],
                  "resume_handle": recovery["resume_handle"]},
        )
        self.assertEqual("303 See Other", resumed["status"])
        self.assertEqual(2, len(provisioning.requests))
        self.assertEqual(
            provisioning.requests[0].idempotency_key,
            provisioning.requests[1].idempotency_key,
        )
        replay, _ = invoke(
            self.app, "/api/v1/public/registrations/resume", method="POST",
            body={"registration_id": recovery["registration_id"],
                  "resume_handle": recovery["resume_handle"]},
        )
        self.assertEqual("403 Forbidden", replay["status"])

    def test_public_registration_duplicate_inputs_never_take_over_identity(self):
        verifier = FakeRegistrationVerification()
        self.app.registration_verification = verifier
        self.app.provisioning = FakeProvisioning()
        self.app.registration_clients = frozenset({"coulomb-social"})
        self.app.registration_tenants = frozenset({"tenant:coulomb"})
        self.app.registration_oidc_issuer = "https://kc.example"
        self.app.registration_password_setup_origins = frozenset({"https://kc.example"})

        def register(username, email):
            invoke(self.app, "/api/v1/public/registrations", method="POST", body={
                "username": username, "email": email,
                "client_id": "coulomb-social", "tenant": "tenant:coulomb",
            })
            registration_id = verifier.requested.registration_id
            verifier.registration_id = registration_id
            result, _ = invoke(
                self.app, "/api/v1/public/registrations/verify", method="POST",
                body={"handle": "x" * 32},
            )
            return result, self.app.service.store.registration_session(registration_id)

        first, first_session = register("first.person", "shared@example.test")
        second, second_session = register("second.person", "shared@example.test")
        collision, collision_session = register("first.person", "other@example.test")

        self.assertEqual("303 See Other", first["status"])
        self.assertEqual("303 See Other", second["status"])
        self.assertNotEqual(first_session.user_id, second_session.user_id)
        self.assertEqual("409 Conflict", collision["status"])
        linked = self.app.service.store.find_identity(
            "https://kc.example", "first.person"
        )
        self.assertEqual(first_session.user_id, linked.user_id)
        self.assertNotEqual(collision_session.user_id, linked.user_id)

    def test_public_registration_rejects_unregistered_client(self):
        self.app.registration_verification = FakeRegistrationVerification()
        self.app.registration_clients = frozenset({"coulomb-social"})
        self.app.registration_tenants = frozenset({"tenant:coulomb"})
        result, _ = invoke(
            self.app,
            "/api/v1/public/registrations",
            method="POST",
            body={
                "username": "person",
                "email": "person@example.test",
                "client_id": "unknown",
                "tenant": "tenant:coulomb",
            },
        )
        self.assertEqual("400 Bad Request", result["status"])

    def test_public_registration_start_is_idempotent_and_key_is_payload_bound(self):
        verifier = FakeRegistrationVerification()
        self.app.registration_verification = verifier
        self.app.registration_clients = frozenset({"coulomb-social"})
        self.app.registration_tenants = frozenset({"tenant:coulomb"})
        body = {
            "username": "idem.person", "email": "idem@example.test",
            "client_id": "coulomb-social", "tenant": "tenant:coulomb",
        }
        headers = {"HTTP_IDEMPOTENCY_KEY": "registration-key-123456789"}
        first, _ = invoke(
            self.app, "/api/v1/public/registrations", method="POST",
            body=body, headers=headers,
        )
        registration_id = verifier.requested.registration_id
        replay, _ = invoke(
            self.app, "/api/v1/public/registrations", method="POST",
            body=body, headers=headers,
        )
        conflicting, _ = invoke(
            self.app, "/api/v1/public/registrations", method="POST",
            body={**body, "email": "different@example.test"}, headers=headers,
        )
        sessions = self.app.service.store.all_registration_sessions()
        self.assertEqual("202 Accepted", first["status"])
        self.assertEqual("202 Accepted", replay["status"])
        self.assertEqual("409 Conflict", conflicting["status"])
        self.assertEqual(1, len(sessions))
        self.assertEqual(1, verifier.request_count)
        self.assertEqual(registration_id, sessions[0].registration_id)
        self.assertNotIn("registration-key-123456789", repr(sessions[0]))

    def test_public_registration_rate_limit_uses_peer_not_forwarded_header(self):
        self.app.registration_verification = FakeRegistrationVerification()
        self.app.registration_rate_limit = 2
        self.app.registration_rate_window_seconds = 60
        body = {
            "username": "person", "email": "person@example.test",
            "client_id": "unknown", "tenant": "tenant:coulomb",
        }
        first, _ = invoke(
            self.app, "/api/v1/public/registrations", method="POST", body=body,
            headers={"HTTP_X_FORWARDED_FOR": "198.51.100.1"},
        )
        second, _ = invoke(
            self.app, "/api/v1/public/registrations", method="POST", body=body,
            headers={"HTTP_X_FORWARDED_FOR": "198.51.100.2"},
        )
        limited, payload = invoke(
            self.app, "/api/v1/public/registrations", method="POST", body=body,
            headers={"HTTP_X_FORWARDED_FOR": "198.51.100.3"},
        )
        other_peer, _ = invoke(
            self.app, "/api/v1/public/registrations", method="POST", body=body,
            remote_addr="127.0.0.2",
        )
        self.assertEqual("400 Bad Request", first["status"])
        self.assertEqual("400 Bad Request", second["status"])
        self.assertEqual("429 Too Many Requests", limited["status"])
        self.assertEqual("rate_limited", json.loads(payload)["error"]["code"])
        self.assertEqual("400 Bad Request", other_peer["status"])

    def test_provision_api_links_provider_subject(self):
        self.app.provisioning = FakeProvisioning()
        created, payload = invoke(
            self.app,
            "/api/v1/tenants/tenant:friendly:binky/users",
            method="POST",
            claims=self.claims,
            body={
                "display_name": "Ada Admin",
                "primary_email": "ada@example.test",
                "role": "tenant-admin",
            },
        )
        self.assertEqual("201 Created", created["status"])
        user_id = json.loads(payload)["user"]["user_id"]
        provisioned, payload = invoke(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/users/{user_id}/provision",
            method="POST",
            claims=self.claims,
        )
        # The helper does not set an idempotency header.
        self.assertEqual("400 Bad Request", provisioned["status"])
        result, payload = invoke_with_idempotency(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/users/{user_id}/provision",
            self.claims,
        )
        self.assertEqual("200 OK", result["status"])
        self.assertEqual("ada", json.loads(payload)["identity"]["subject"])
        changed, payload = invoke_with_idempotency(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/users/{user_id}",
            self.claims,
            method="PATCH",
            body={"status": "suspended"},
        )
        self.assertEqual("200 OK", changed["status"])
        self.assertEqual("suspended", json.loads(payload)["status"])
        self.assertIn(("suspend", "ada"), self.app.provisioning.actions)
        removed, payload = invoke_with_idempotency(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/users/{user_id}",
            self.claims, method="DELETE",
        )
        self.assertEqual("200 OK", removed["status"])
        self.assertEqual("removed", json.loads(payload)["status"])
        self.assertIn(("deprovision", "ada"), self.app.provisioning.actions)

    def test_invitation_lifecycle_is_versioned_and_replay_safe(self):
        created, payload = invoke(
            self.app,
            "/api/v1/tenants/tenant:friendly:binky/invitations",
            method="POST", claims=self.claims,
            body={"primary_email": "invitee@example.test", "role": "user"},
        )
        self.assertEqual("201 Created", created["status"])
        invitation = json.loads(payload)["invitation"]
        invitation_id = invitation["invitation_id"]
        self.assertEqual(1, invitation["version"])
        self.assertIsNotNone(invitation["expires_at"])

        duplicate, _ = invoke(
            self.app,
            "/api/v1/tenants/tenant:friendly:binky/invitations",
            method="POST", claims=self.claims,
            body={"primary_email": "INVITEE@example.test", "role": "user"},
        )
        self.assertEqual("409 Conflict", duplicate["status"])

        listed, payload = invoke(
            self.app, "/api/v1/tenants/tenant:friendly:binky/invitations",
            claims=self.claims,
        )
        self.assertEqual("200 OK", listed["status"])
        self.assertEqual(1, len(json.loads(payload)["items"]))

        resent, payload = invoke(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/invitations/{invitation_id}/resend",
            method="POST", claims=self.claims, headers={"HTTP_IF_MATCH": '"1"'},
        )
        self.assertEqual("200 OK", resent["status"])
        self.assertEqual(2, json.loads(payload)["version"])
        conflict, _ = invoke(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/invitations/{invitation_id}/expire",
            method="POST", claims=self.claims, headers={"HTTP_IF_MATCH": '"1"'},
        )
        self.assertEqual("409 Conflict", conflict["status"])
        expired, payload = invoke(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/invitations/{invitation_id}/expire",
            method="POST", claims=self.claims, headers={"HTTP_IF_MATCH": '"2"'},
        )
        self.assertEqual("200 OK", expired["status"])
        self.assertEqual("revoked", json.loads(payload)["status"])

        claimable, payload = invoke(
            self.app, "/api/v1/tenants/tenant:friendly:binky/invitations",
            method="POST", claims=self.claims,
            body={"primary_email": "sample.user@example.test", "role": "user"},
        )
        self.assertEqual("201 Created", claimable["status"])
        claim_id = json.loads(payload)["invitation"]["invitation_id"]
        claimed, _ = invoke(
            self.app, f"/api/v1/invitations/{claim_id}/claim",
            method="POST", claims=self.claims,
        )
        self.assertEqual("200 OK", claimed["status"])
        replayed, _ = invoke(
            self.app, f"/api/v1/invitations/{claim_id}/claim",
            method="POST", claims=self.claims,
        )
        self.assertEqual("400 Bad Request", replayed["status"])

        timed, payload = invoke(
            self.app, "/api/v1/tenants/tenant:friendly:binky/invitations",
            method="POST", claims=self.claims,
            body={"primary_email": "expired@example.test", "role": "user"},
        )
        timed_id = json.loads(payload)["invitation"]["invitation_id"]
        timed_invitation = self.app.service.store.family_invitation(timed_id)
        self.app.service.store.save_family_invitation(replace(
            timed_invitation, expires_at=utc_now() - timedelta(seconds=1)
        ))
        expired_claims = dict(self.claims)
        expired_claims["email"] = "expired@example.test"
        timed_out, _ = invoke(
            self.app, f"/api/v1/invitations/{timed_id}/claim",
            method="POST", claims=expired_claims,
        )
        self.assertEqual("400 Bad Request", timed_out["status"])

    def test_outbox_delivery_replay_and_dead_letter(self):
        invoke(self.app, "/api/v1/me", claims=self.claims)
        actor = self.app.service.identity_adapter.normalize(self.claims)
        event = self.app.service.outbox_events()[0]

        def unavailable(_event):
            raise RuntimeError("notification provider unavailable; token=redacted")

        first = self.app.service.deliver_outbox(
            actor, unavailable, worker_id="worker-a", max_attempts=2
        )[0]
        self.assertEqual(1, first.delivery_attempts)
        self.assertIsNotNone(first.failed_at)
        self.assertIsNone(first.dead_lettered_at)
        replayed = self.app.service.replay_outbox(actor, event.event_id)
        self.assertIsNone(replayed.failure_reason)
        second = self.app.service.deliver_outbox(
            actor, unavailable, worker_id="worker-a", max_attempts=2
        )[0]
        self.assertIsNotNone(second.dead_lettered_at)
        self.assertLessEqual(len(second.failure_reason), 200)

        replayed = self.app.service.replay_outbox(actor, event.event_id)
        delivered = self.app.service.deliver_outbox(
            actor, lambda _event: None, worker_id="worker-b"
        )[0]
        self.assertEqual(replayed.event_id, delivered.event_id)
        self.assertIsNotNone(delivered.delivered_at)
        self.assertNotIn(delivered, self.app.service.outbox_events())

    def test_platform_tenant_bootstrap_recovery_and_outbox_transport(self):
        self.app.tenant_management = FakeTenantManagement()
        self.app.provisioning = FakeProvisioning()
        claims = self.platform_claims()
        denied, _ = invoke_with_idempotency(
            self.app, "/api/v1/platform/tenants", self.claims,
            body={"tenant": "tenant:friendly:new", "display_name": "New"},
        )
        self.assertEqual("403 Forbidden", denied["status"])
        created, payload = invoke_with_idempotency(
            self.app, "/api/v1/platform/tenants", claims,
            body={
                "tenant": "tenant:friendly:new", "display_name": "New Tenant",
                "first_admin": {"primary_email": "admin@new.test", "display_name": "Admin"},
            },
        )
        self.assertEqual("201 Created", created["status"])
        decoded = json.loads(payload)
        self.assertEqual("created", decoded["tenant"]["status"])
        self.assertEqual("tenant-admin", decoded["first_admin"]["membership"]["kind"])
        user_id = decoded["first_admin"]["user"]["user_id"]
        recovered, payload = invoke_with_idempotency(
            self.app,
            f"/api/v1/platform/tenants/tenant:friendly:new/users/{user_id}/recover",
            claims,
        )
        self.assertEqual("200 OK", recovered["status"])
        self.assertEqual("active", json.loads(payload)["tenant_account"]["status"])

        self.app.outbox_delivery = lambda _event: None
        delivered, payload = invoke(
            self.app, "/api/v1/platform/outbox/deliver", method="POST",
            claims=claims, body={"worker_id": "test-worker"},
        )
        self.assertEqual("200 OK", delivered["status"])
        self.assertTrue(json.loads(payload)["items"])

    def test_platform_tenant_lifecycle_is_delegated_to_the_authority(self):
        authority = FakeTenantManagement()
        self.app.tenant_management = authority
        claims = self.platform_claims()
        tenant = "tenant:friendly:lifecycle"
        invoke_with_idempotency(
            self.app, "/api/v1/platform/tenants", claims,
            body={"tenant": tenant, "display_name": "Lifecycle"},
        )
        path = f"/api/v1/platform/tenants/{tenant}"

        denied, _ = invoke(self.app, path, claims=self.claims)
        self.assertEqual("403 Forbidden", denied["status"])

        read, payload = invoke(self.app, path, claims=claims)
        self.assertEqual("200 OK", read["status"])
        record = json.loads(payload)
        self.assertEqual("active", record["lifecycle"])
        self.assertEqual(1, record["version"])

        def mutate(suffix, *, method, body, version, key):
            return invoke(
                self.app, path + suffix, method=method, claims=claims, body=body,
                headers={
                    "HTTP_IF_MATCH": f'"{version}"',
                    "HTTP_IDEMPOTENCY_KEY": key,
                },
            )

        unconditional, _ = invoke(
            self.app, path, method="PATCH", claims=claims,
            body={"metadata": {"display_name": "X"}, "reason": "rename"},
            headers={"HTTP_IDEMPOTENCY_KEY": "tenant-update-0000000000"},
        )
        self.assertEqual("400 Bad Request", unconditional["status"])

        unreasoned, _ = mutate(
            "", method="PATCH", body={"metadata": {"display_name": "X"}},
            version=1, key="tenant-update-0000000001",
        )
        self.assertEqual("400 Bad Request", unreasoned["status"])

        updated, payload = mutate(
            "", method="PATCH",
            body={"metadata": {"display_name": "Renamed"}, "reason": "operator rename"},
            version=1, key="tenant-update-0000000002",
        )
        self.assertEqual("200 OK", updated["status"])
        self.assertEqual("Renamed", json.loads(payload)["display_name"])
        self.assertEqual(2, json.loads(payload)["version"])

        replayed, payload = mutate(
            "", method="PATCH",
            body={"metadata": {"display_name": "Renamed"}, "reason": "operator rename"},
            version=1, key="tenant-update-0000000002",
        )
        self.assertEqual("200 OK", replayed["status"])
        self.assertTrue(json.loads(payload)["replayed"])
        self.assertEqual(2, json.loads(payload)["version"])

        stale, _ = mutate(
            "", method="PATCH",
            body={"metadata": {"display_name": "Again"}, "reason": "second rename"},
            version=1, key="tenant-update-0000000003",
        )
        self.assertEqual("409 Conflict", stale["status"])

        retired, payload = mutate(
            "/retire", method="POST", body={"reason": "contract ended"},
            version=2, key="tenant-retire-0000000001",
        )
        self.assertEqual("200 OK", retired["status"])
        self.assertEqual("retired", json.loads(payload)["lifecycle"])

        while_retired, _ = mutate(
            "", method="PATCH",
            body={"metadata": {"display_name": "Nope"}, "reason": "late rename"},
            version=3, key="tenant-update-0000000004",
        )
        self.assertEqual("409 Conflict", while_retired["status"])

        double, _ = mutate(
            "/retire", method="POST", body={"reason": "again"},
            version=3, key="tenant-retire-0000000002",
        )
        self.assertEqual("409 Conflict", double["status"])

        reactivated, payload = mutate(
            "/reactivate", method="POST", body={"reason": "contract renewed"},
            version=3, key="tenant-reactivate-000001",
        )
        self.assertEqual("200 OK", reactivated["status"])
        self.assertEqual("active", json.loads(payload)["lifecycle"])

        # user-engine keeps no tenant table of its own: every read and write
        # above went to the authority.
        self.assertEqual({tenant}, set(authority.records))

        missing, _ = invoke(
            self.app, "/api/v1/platform/tenants/tenant:friendly:absent/retire",
            method="POST", claims=claims, body={"reason": "unknown"},
            headers={
                "HTTP_IF_MATCH": '"1"',
                "HTTP_IDEMPOTENCY_KEY": "tenant-retire-0000000009",
            },
        )
        self.assertEqual("404 Not Found", missing["status"])

    def test_platform_browser_tenant_lifecycle_controls(self):
        authority = FakeTenantManagement()
        claims = self.platform_claims()
        oidc = OIDCClient(
            issuer="https://kc.example", client_id="portal",
            redirect_uri="https://users.example/oidc/callback", audience="portal",
        )
        oidc.sessions["platform"] = BrowserSession(
            claims=claims, expires_at=9999999999, csrf_token="platform-csrf",
        )
        self.app.oidc_client = oidc
        self.app.tenant_management = authority
        tenant = "tenant:friendly:browserlifecycle"
        authority.create_tenant(
            tenant=tenant, display_name="Browser Lifecycle",
            idempotency_key="seed", correlation_id="corr",
        )
        quoted = quote(tenant, safe="")

        lookup, _ = invoke(
            self.app, "/platform/tenant", cookie="ue_session=platform",
            query=urlencode({"tenant": tenant}),
        )
        self.assertEqual("303 See Other", lookup["status"])
        self.assertEqual(f"/platform/tenants/{quoted}", dict(lookup["headers"])["Location"])

        page, html = invoke(
            self.app, f"/platform/tenants/{quoted}", cookie="ue_session=platform"
        )
        self.assertEqual("200 OK", page["status"])
        self.assertIn(b"Retire tenant", html)
        self.assertIn(b"not reported", html)
        self.assertIn(b'name="version" value="1"', html)

        forged, _ = invoke(
            self.app, f"/platform/tenants/{quoted}", method="POST",
            cookie="ue_session=platform", form={
                "csrf_token": "wrong", "operation": "retire",
                "version": "1", "reason": "forged",
            },
        )
        self.assertEqual("403 Forbidden", forged["status"])
        self.assertEqual("active", authority.records[tenant].lifecycle)

        renamed, html = invoke(
            self.app, f"/platform/tenants/{quoted}", method="POST",
            cookie="ue_session=platform", form={
                "csrf_token": "platform-csrf", "operation": "update", "version": "1",
                "display_name": "Renamed In Browser", "reason": "operator rename",
            },
        )
        self.assertEqual("200 OK", renamed["status"])
        self.assertIn(b"Renamed In Browser", html)

        retired, html = invoke(
            self.app, f"/platform/tenants/{quoted}", method="POST",
            cookie="ue_session=platform", form={
                "csrf_token": "platform-csrf", "operation": "retire",
                "version": "2", "reason": "contract ended",
            },
        )
        self.assertEqual("200 OK", retired["status"])
        self.assertIn(b"Reactivate tenant", html)
        # A retired tenant offers no metadata form, matching the authority.
        self.assertNotIn(b"Save metadata", html)

        resubmitted, html = invoke(
            self.app, f"/platform/tenants/{quoted}", method="POST",
            cookie="ue_session=platform", form={
                "csrf_token": "platform-csrf", "operation": "retire",
                "version": "2", "reason": "contract ended",
            },
        )
        self.assertEqual("200 OK", resubmitted["status"])
        self.assertIn(b"replayed", html)
        self.assertEqual(3, authority.records[tenant].version)

        platform, html = invoke(
            self.app, "/platform", cookie="ue_session=platform"
        )
        self.assertIn(b"Manage an existing tenant", html)

    def test_platform_tenant_authority_denial_is_redacted_and_creates_no_admin(self):
        oidc = OIDCClient(
            issuer="https://kc.example", client_id="portal",
            redirect_uri="https://users.example/oidc/callback", audience="portal",
        )
        oidc.sessions["platform"] = BrowserSession(
            claims=self.platform_claims(), expires_at=9999999999,
            csrf_token="platform-csrf",
        )
        self.app.oidc_client = oidc
        self.app.tenant_management = HTTPTenantManagementAdapter(
            base_url="https://tenants.example", bearer_token="test-service-boundary",
        )
        failure = HTTPError(
            "https://tenants.example/tenants", 403, "Forbidden", {},
            io.BytesIO(b'{"error_code":"write_denied","detail":"private-policy-detail"}'),
        )
        with patch("user_engine.adapters.tenant_management.urlopen", side_effect=failure), \
                patch.object(self.app.service, "create_user") as create_user:
            response, payload = invoke(
                self.app, "/platform/tenants", method="POST",
                cookie="ue_session=platform", form={
                    "csrf_token": "platform-csrf", "tenant": "tenant:trial:demo-company",
                    "admin_display_name": "First Admin", "admin_email": "admin@example.test",
                },
            )
        self.assertEqual("403 Forbidden", response["status"])
        self.assertNotIn(b"provisioning_unavailable", payload)
        self.assertNotIn(b"private-policy-detail", payload)
        create_user.assert_not_called()

    def test_platform_browser_tenant_and_first_admin_bootstrap(self):
        claims = self.platform_claims()
        oidc = OIDCClient(
            issuer="https://kc.example", client_id="portal",
            redirect_uri="https://users.example/oidc/callback", audience="portal",
        )
        oidc.sessions["platform"] = BrowserSession(
            claims=claims, expires_at=9999999999, csrf_token="platform-csrf",
        )
        self.app.oidc_client = oidc
        self.app.tenant_management = FakeTenantManagement()
        self.app.provisioning = FakeProvisioning()
        page, html = invoke(self.app, "/platform", cookie="ue_session=platform")
        self.assertEqual("200 OK", page["status"])
        self.assertIn(b"First administrator", html)
        denied, _ = invoke(
            self.app, "/platform/tenants", method="POST",
            cookie="ue_session=platform", form={
                "csrf_token": "wrong", "tenant": "tenant:friendly:browser",
                "display_name": "Browser Tenant",
            },
        )
        self.assertEqual("403 Forbidden", denied["status"])
        created, html = invoke(
            self.app, "/platform/tenants", method="POST",
            cookie="ue_session=platform", form={
                "csrf_token": "platform-csrf", "tenant": "tenant:friendly:browser",
                "display_name": "Browser Tenant", "admin_display_name": "First Admin",
                "admin_email": "first-admin@browser.test",
            },
        )
        self.assertEqual("200 OK", created["status"])
        self.assertIn(b"awaiting onboarding", html)
        memberships = self.app.service.store.memberships_for_tenant(
            "tenant:friendly:browser"
        )
        self.assertEqual("tenant-admin", memberships[0].kind)
        admin_page, html = invoke(
            self.app, "/admin/tenant:friendly:browser", cookie="ue_session=platform"
        )
        self.assertEqual("200 OK", admin_page["status"])
        self.assertIn(b"Lifecycle diagnostics", html)
        self.assertIn(b"Recover identity", html)
        recovered, _ = invoke(
            self.app,
            f"/admin/tenant:friendly:browser/users/{memberships[0].user_id}/recover",
            method="POST", cookie="ue_session=platform",
            form={"csrf_token": "platform-csrf"},
        )
        self.assertEqual("303 See Other", recovered["status"])

    def test_admin_form_requires_csrf_and_supports_two_step_provisioning(self):
        oidc = OIDCClient(
            issuer="https://kc.example",
            client_id="portal",
            redirect_uri="https://users.example/oidc/callback",
            audience="portal",
        )
        oidc.sessions["browser"] = BrowserSession(
            claims=self.claims,
            expires_at=9999999999,
            csrf_token="csrf-test-token",
        )
        self.app.oidc_client = oidc
        self.app.provisioning = FakeProvisioning()
        denied, _ = invoke(
            self.app,
            "/admin/tenant:friendly:binky/users",
            method="POST",
            cookie="ue_session=browser",
            form={
                "csrf_token": "wrong",
                "display_name": "Ada Admin",
                "primary_email": "ada@example.test",
                "role": "tenant-admin",
            },
        )
        self.assertEqual("403 Forbidden", denied["status"])
        created, _ = invoke(
            self.app,
            "/admin/tenant:friendly:binky/users",
            method="POST",
            cookie="ue_session=browser",
            form={
                "csrf_token": "csrf-test-token",
                "display_name": "Ada Admin",
                "primary_email": "ada@example.test",
                "role": "tenant-admin",
            },
        )
        self.assertEqual("303 See Other", created["status"])
        page, html = invoke(
            self.app,
            "/admin/tenant:friendly:binky",
            cookie="ue_session=browser",
        )
        self.assertEqual("200 OK", page["status"])
        self.assertIn(b"ada@example.test", html)
        self.assertIn(b"Create login", html)
        user_id = next(iter(self.app.service.store.users))
        handoff, html = invoke(
            self.app,
            f"/admin/tenant:friendly:binky/users/{user_id}/provision",
            method="POST",
            cookie="ue_session=browser",
            form={"csrf_token": "csrf-test-token"},
        )
        self.assertEqual("200 OK", handoff["status"])
        self.assertIn(b"Continue to password setup", html)
        self.assertIn(b"https://kc.example/setup/password?token=opaque", html)
        page, html = invoke(
            self.app,
            "/admin/tenant:friendly:binky",
            cookie="ue_session=browser",
        )
        self.assertEqual("200 OK", page["status"])
        self.assertIn(b"Create password setup link", html)
        self.assertIn(b"Invite a user", html)
        self.assertIn(b"Remove account", html)

    def test_browser_invitation_acceptance_and_onboarding_status(self):
        oidc = OIDCClient(
            issuer="https://kc.example", client_id="portal",
            redirect_uri="https://users.example/oidc/callback", audience="portal",
        )
        oidc.sessions["browser"] = BrowserSession(
            claims=self.claims, expires_at=9999999999, csrf_token="csrf-test-token",
        )
        self.app.oidc_client = oidc
        created, _ = invoke(
            self.app, "/admin/tenant:friendly:binky/invitations", method="POST",
            cookie="ue_session=browser", form={
                "csrf_token": "csrf-test-token", "display_name": "Invitee",
                "primary_email": "invitee@example.test", "role": "user",
            },
        )
        self.assertEqual("303 See Other", created["status"])
        invitation = next(iter(self.app.service.store.family_invitations.values()))
        page, html = invoke(
            self.app, f"/invitations/{invitation.invitation_id}",
            cookie="ue_session=browser",
        )
        self.assertEqual("200 OK", page["status"])
        self.assertIn(b"Accept invitation", html)

        wrong, _ = invoke(
            self.app, f"/invitations/{invitation.invitation_id}", method="POST",
            cookie="ue_session=browser", form={"csrf_token": "csrf-test-token"},
        )
        self.assertEqual("403 Forbidden", wrong["status"])
        invitee_claims = human_actor_claims(
            subject="invitee", tenant="tenant:friendly:binky"
        )
        invitee_claims["email"] = "invitee@example.test"
        oidc.sessions["browser"] = BrowserSession(
            claims=invitee_claims, expires_at=9999999999,
            csrf_token="csrf-test-token",
        )
        accepted, _ = invoke(
            self.app, f"/invitations/{invitation.invitation_id}", method="POST",
            cookie="ue_session=browser", form={"csrf_token": "csrf-test-token"},
        )
        self.assertEqual("303 See Other", accepted["status"])
        self.assertEqual("/onboarding", accepted["headers"]["Location"])
        own_journey = OnboardingJourney(
            tenant="tenant:friendly:binky", user_id=invitation.user_id,
            protocol_id="protocol-self", trigger_type=OnboardingTriggerType.INVITATION,
            status=OnboardingJourneyStatus.IN_PROGRESS, active_step_key="profile-review",
            steps=(OnboardingStep(
                step_key="profile-review", title="Review your profile",
                subsystem="user-engine", status=OnboardingStepStatus.IN_PROGRESS,
            ),),
        )
        provider_journey = OnboardingJourney(
            tenant="tenant:friendly:binky", user_id=invitation.user_id,
            protocol_id="protocol-provider", trigger_type=OnboardingTriggerType.INVITATION,
            status=OnboardingJourneyStatus.BLOCKED, active_step_key="mfa",
            steps=(OnboardingStep(
                step_key="mfa", title="Enroll MFA", subsystem="key-cape",
                status=OnboardingStepStatus.BLOCKED,
                handoff=SubsystemHandoff(
                    subsystem="key-cape", status=OnboardingStepStatus.BLOCKED,
                ),
            ),),
        )
        self.app.service.store.save_onboarding_journey(own_journey)
        self.app.service.store.save_onboarding_journey(provider_journey)
        onboarding, html = invoke(
            self.app, "/onboarding", cookie="ue_session=browser"
        )
        self.assertEqual("200 OK", onboarding["status"])
        self.assertIn(b"Onboarding progress", html)
        self.assertIn(b"tenant:friendly:binky", html)
        self.assertIn(b"Profile and consent", html)
        self.assertIn(b"Mark complete", html)
        self.assertIn(b"provider-owned surface", html)
        completed, _ = invoke(
            self.app,
            f"/onboarding/{own_journey.journey_id}/steps/profile-review/complete",
            method="POST", cookie="ue_session=browser",
            form={"csrf_token": "csrf-test-token"},
        )
        self.assertEqual("303 See Other", completed["status"])
        provider_denied, _ = invoke(
            self.app,
            f"/onboarding/{provider_journey.journey_id}/steps/mfa/complete",
            method="POST", cookie="ue_session=browser",
            form={"csrf_token": "csrf-test-token"},
        )
        self.assertEqual("403 Forbidden", provider_denied["status"])
        saved, _ = invoke(
            self.app, "/onboarding/profile", method="POST",
            cookie="ue_session=browser", form={
                "csrf_token": "csrf-test-token", "display_name": "Updated Invitee",
                "consent_accepted": "yes",
            },
        )
        self.assertEqual("303 See Other", saved["status"])
        user = self.app.service.store.user(invitation.user_id)
        self.assertEqual("Updated Invitee", user.display_name)
        self.assertEqual("invitee@example.test", user.primary_email)
        self.assertEqual("portal-terms-v1", user.consent_version)
        self.assertIsNotNone(user.profile_completed_at)


class FakeRegistrationVerification:
    def __init__(self):
        self.requested = None
        self.registration_id = None
        self.request_count = 0

    def request(self, request):
        self.request_count += 1
        self.requested = request
        return RegistrationVerificationReceipt(request_id="vrq_test")

    def consume(self, opaque_handle):
        return VerifiedRegistrationApplicant(
            verification_id="fvr_test",
            registration_id=self.registration_id,
            normalized_email=self.requested.normalized_email,
            preferred_username=self.requested.preferred_username,
            client_id=self.requested.client_id,
            tenant=self.requested.tenant,
            source_system="mail-verifier",
            assurance={"mailbox_control": True},
            display_name=self.requested.display_name,
        )

    def cancel(self, opaque_handle):
        return self.consume(opaque_handle)


class FakeProvisioning:
    def __init__(self, password_setup_url="https://kc.example/setup/password?token=opaque"):
        self.actions = []
        self.requests = []
        self.password_setup_url = password_setup_url

    def provision(self, request):
        self.requests.append(request)
        self.actions.append(("provision", request.primary_email))
        return ProvisioningResult(
            provider="netkingdom-lldap",
            external_subject=request.preferred_username or request.primary_email.split("@")[0],
            status="password_setup_required",
            password_setup_url=self.password_setup_url,
        )

    def suspend(self, *, external_subject, idempotency_key, correlation_id):
        self.actions.append(("suspend", external_subject))
        return ProvisioningResult("netkingdom-lldap", external_subject, "suspended")

    def reactivate(self, *, external_subject, idempotency_key, correlation_id):
        self.actions.append(("reactivate", external_subject))
        return ProvisioningResult("netkingdom-lldap", external_subject, "active")

    def deprovision(self, *, external_subject, idempotency_key, correlation_id):
        self.actions.append(("deprovision", external_subject))
        return ProvisioningResult("netkingdom-lldap", external_subject, "removed")

    def reconcile(self, request, *, external_subject, desired_status="active"):
        self.actions.append(("reconcile", external_subject))
        return IdentityDriftResult(
            "netkingdom-lldap", external_subject, "in_sync", changed=("status",)
        )


class FailingOnceProvisioning(FakeProvisioning):
    def provision(self, request):
        self.requests.append(request)
        if len(self.requests) == 1:
            raise RuntimeError("provider unavailable")
        return ProvisioningResult(
            provider="netkingdom-lldap",
            external_subject=request.preferred_username,
            status="password_setup_required",
            password_setup_url=self.password_setup_url,
        )


class FakeTenantManagement:
    """Stands in for tenant-engine, including its compare-and-swap semantics."""

    def __init__(self):
        self.records = {}
        self.receipts = {}
        self.reasons = []

    def create_tenant(self, *, tenant, display_name, idempotency_key, correlation_id):
        self.records.setdefault(tenant, TenantRecord(
            tenant=tenant, external_ref=tenant, lifecycle="active", version=1,
            display_name=display_name,
        ))
        return TenantProvisioningResult(tenant=tenant, status="created")

    def tenant(self, *, tenant, correlation_id):
        record = self.records.get(tenant)
        if record is None:
            raise NotFoundError("tenant not found")
        return record

    def update_tenant(self, *, tenant, metadata, expected_version, reason,
                      idempotency_key, correlation_id):
        record = self._mutate(tenant, expected_version, reason, idempotency_key)
        if record is not None:
            return record
        current = self.records[tenant]
        if current.lifecycle == "retired":
            raise ConflictError("invalid_lifecycle_transition")
        return self._commit(idempotency_key, replace(
            current, version=current.version + 1,
            display_name=metadata.get("display_name", current.display_name),
            contact_email=metadata.get("contact_email", current.contact_email),
        ))

    def retire_tenant(self, *, tenant, expected_version, reason, idempotency_key,
                      correlation_id):
        return self._transition(
            tenant, "retired", expected_version, reason, idempotency_key
        )

    def reactivate_tenant(self, *, tenant, expected_version, reason, idempotency_key,
                          correlation_id):
        return self._transition(
            tenant, "active", expected_version, reason, idempotency_key
        )

    def _transition(self, tenant, lifecycle, expected_version, reason, idempotency_key):
        record = self._mutate(tenant, expected_version, reason, idempotency_key)
        if record is not None:
            return record
        current = self.records[tenant]
        if current.lifecycle == lifecycle:
            raise ConflictError("invalid_lifecycle_transition")
        return self._commit(idempotency_key, replace(
            current, lifecycle=lifecycle, version=current.version + 1
        ))

    def _mutate(self, tenant, expected_version, reason, idempotency_key):
        if idempotency_key in self.receipts:
            return replace(self.receipts[idempotency_key], replayed=True)
        if tenant not in self.records:
            raise NotFoundError("tenant not found")
        if not reason:
            raise ValidationError("invalid_update")
        if expected_version != self.records[tenant].version:
            raise ConflictError("version_conflict")
        self.reasons.append(reason)
        return None

    def _commit(self, idempotency_key, record):
        self.records[record.tenant] = record
        self.receipts[idempotency_key] = record
        return record


class FailingProvisioning(FakeProvisioning):
    def provision(self, request):
        raise RuntimeError("provider-secret must never escape")


def invoke_with_idempotency(app, path, claims, *, method="POST", body=None):
    payload = json.dumps(body or {}).encode()
    environ = {
        "REQUEST_METHOD": method,
        "PATH_INFO": path,
        "QUERY_STRING": "",
        "CONTENT_LENGTH": str(len(payload)),
        "wsgi.input": io.BytesIO(payload),
        "HTTP_X_REQUEST_ID": "corr_test",
        "HTTP_X_VERIFIED_OIDC_CLAIMS": json.dumps(claims),
        "HTTP_X_USER_ENGINE_PROXY_SECRET": SECRET,
        "HTTP_IDEMPOTENCY_KEY": "test-idempotency-123456",
    }
    captured = {}
    response = b"".join(app(
        environ,
        lambda status, headers: captured.update(
            {"status": status, "headers": dict(headers)}
        ),
    ))
    return captured, response


if __name__ == "__main__":
    unittest.main()