Delegate tenant lifecycle to the tenant authority

TEN-WP-0005 landed the authoritative metadata update and reversible
retirement contract, so USER-WP-0021-T01's deferred tenant operations are
now implementable without user-engine inventing lifecycle semantics.

TenantManagementPort gains read, update, retire, and reactivate. The HTTP
adapter echoes the record version as an If-Match ETag (never `*`), sends an
Idempotency-Key plus actor/reason/correlation_id, and surfaces
Idempotent-Replay. Authority failures map to redacted domain errors carrying
only the contract's stable error_code; its detail text never crosses the
boundary.

Platform operators get the matching API routes and a CSRF-protected browser
screen that reads the record before mutating it and hides the metadata form
for a retired tenant. Portal OpenAPI moves to 0.3.0 with TenantRecord,
UpdateTenant, and TenantLifecycleChange.

Full suite: 145 tests, 3 external-provider skips.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

tests/test_web.py

@@ -5,7 +5,7 @@ import re
 import unittest
 from dataclasses import replace
 from datetime import timedelta
-from urllib.parse import urlencode
+from urllib.parse import quote, urlencode
 
 from user_engine.adapters import InMemoryUserEngineStore, LocalAuthorizationCheckPort
 from user_engine.domain import (
@@ -18,8 +18,10 @@ from user_engine.ports import (
     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
@@ -706,6 +708,195 @@ class PortalApplicationTests(unittest.TestCase):
         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'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(
@@ -1008,9 +1199,80 @@ class FailingOnceProvisioning(FakeProvisioning):
 
 
 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):