Make public registration provisioning resumable

src/user_engine/domain/models.py

@@ -579,6 +579,7 @@ class RegistrationSession:
     started_by_subject: str | None = None
     applicant_username: str | None = None
     client_id: str | None = None
+    provisioning_resume_hash: str | None = None
     correlation_id: str | None = None
     created_at: datetime = field(default_factory=utc_now)
     updated_at: datetime = field(default_factory=utc_now)

src/user_engine/web.py

@@ -9,9 +9,11 @@ of user-engine.
 
 from __future__ import annotations
 
-from dataclasses import asdict, is_dataclass
+from dataclasses import asdict, is_dataclass, replace
 from enum import Enum
 from html import escape
+import hashlib
+import hmac
 import json
 import re
 import secrets
@@ -177,6 +179,10 @@ class PortalApplication:
             return self._verify_public_registration(
                 environ, start_response, correlation_id
             )
+        if path == "/api/v1/public/registrations/resume" and method == "POST":
+            return self._resume_public_registration(
+                environ, start_response, correlation_id
+            )
 
         actor = self._actor(environ)
         if path == "/api/v1/me" and method == "GET":
@@ -907,6 +913,12 @@ class PortalApplication:
             ),
             correlation_id=correlation_id,
         )
+        resume_handle = secrets.token_urlsafe(32)
+        updated = replace(
+            updated,
+            provisioning_resume_hash=hashlib.sha256(resume_handle.encode()).hexdigest(),
+        )
+        self.service.store.save_registration_session(updated)
         if self.provisioning is None or not self.registration_oidc_issuer:
             raise RuntimeError("public registration provisioning is unavailable")
         completion = self.service.complete_registration(
@@ -916,26 +928,77 @@ class PortalApplication:
             primary_email=evidence.normalized_email,
             correlation_id=correlation_id,
         )
+        try:
+            return self._provision_public_registration(
+                start_response, actor, completion.user, completion.session,
+                evidence.normalized_email, evidence.display_name,
+                evidence.preferred_username, correlation_id,
+            )
+        except RuntimeError:
+            return self._json(start_response, "202 Accepted", {
+                "status": "provisioning_pending",
+                "registration_id": updated.registration_id,
+                "resume_handle": resume_handle,
+            }, correlation_id)
+
+    def _resume_public_registration(self, environ, start_response, correlation_id):
+        if not self.public_registration:
+            raise NotFoundError("public registration is unavailable")
+        body = self._body(environ)
+        session = self.service.store.registration_session(str(body.get("registration_id") or ""))
+        handle = str(body.get("resume_handle") or "")
+        digest = hashlib.sha256(handle.encode()).hexdigest()
+        if (
+            session is None or session.status.value != "completed"
+            or not session.provisioning_resume_hash
+            or not hmac.compare_digest(digest, session.provisioning_resume_hash)
+        ):
+            raise AuthorizationDenied("registration resume is invalid")
+        factors = self.service.store.factors_for_registration(session.registration_id)
+        email_factor = next((item for item in factors if item.factor_type == IdentityFactorType.EMAIL), None)
+        if email_factor is None or not session.user_id:
+            raise ValidationError("registration is not recoverable")
+        actor = Actor(
+            issuer="urn:netkingdom:public-registration", subject=str(session.started_by_subject),
+            tenant=session.tenant, principal_type=PrincipalType.HUMAN,
+            audience=("user-engine",), roles=("registration-applicant",),
+            authorized_party=session.client_id, preferred_username=session.applicant_username,
+        )
+        user = self.service.store.user(session.user_id)
+        if user is None:
+            raise ValidationError("registration user is unavailable")
+        return self._provision_public_registration(
+            start_response, actor, user, session, email_factor.normalized_value,
+            user.display_name, str(session.applicant_username), correlation_id,
+        )
+
+    def _provision_public_registration(
+        self, start_response, actor, user, session, email, display_name,
+        preferred_username, correlation_id,
+    ):
         provisioned = self.provisioning.provision(
             ProvisioningRequest(
-                user_id=completion.user.user_id,
-                tenant=updated.tenant,
-                primary_email=evidence.normalized_email,
-                display_name=evidence.display_name,
-                preferred_username=evidence.preferred_username,
-                idempotency_key=f"public-registration-{updated.registration_id}",
+                user_id=user.user_id,
+                tenant=session.tenant,
+                primary_email=email,
+                display_name=display_name,
+                preferred_username=preferred_username,
+                idempotency_key=f"public-registration-{session.registration_id}",
                 correlation_id=correlation_id,
                 roles=("user",),
             )
         )
         self.service.link_identity(
             actor,
-            completion.user.user_id,
+            user.user_id,
             issuer=self.registration_oidc_issuer,
             subject=provisioned.external_subject,
             provider=provisioned.provider,
             correlation_id=correlation_id,
         )
+        self.service.store.save_registration_session(
+            replace(session, provisioning_resume_hash=None)
+        )
         if provisioned.password_setup_url:
             self._validate_registration_handoff(provisioned.password_setup_url)
             return self._redirect(

tests/test_web.py

@@ -271,6 +271,44 @@ class PortalApplicationTests(unittest.TestCase):
         )
         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_rejects_unregistered_client(self):
         self.app.registration_verification = FakeRegistrationVerification()
         self.app.registration_clients = frozenset({"coulomb-social"})
@@ -772,6 +810,19 @@ class FakeProvisioning:
         )
 
 
+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:
     def create_tenant(self, *, tenant, display_name, idempotency_key, correlation_id):
         return TenantProvisioningResult(tenant=tenant, status="created")

workplans/USER-WP-0022-public-registration-and-jit-application-profiles.md

@@ -90,6 +90,13 @@ the provider-owned password-setup handoff. `ProvisioningRequest` now carries
 an optional canonical `preferred_username` through the HTTP adapter without
 changing existing callers.
 
+Retry-safe recovery is now implemented for the local-completion/provider-
+failure window. A single-use mailbox handle is never replayed; instead,
+user-engine returns a separate random resume handle, stores only its SHA-256
+digest, retries with the same registration idempotency key and local user,
+and invalidates the handle after provider linking. Replay is denied. The full
+suite passes 127 tests with 3 environment-dependent skips.
+
 ## T03 - Create application profiles on first login
 
 ```task