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 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_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()