Implement registration identity model

2026-06-15 22:06:39 +02:00
parent 2c94b40fc4
commit a36a25898e
12 changed files with 1012 additions and 12 deletions
--- a/src/user_engine/adapters/local.py
+++ b/src/user_engine/adapters/local.py
@@ -18,10 +18,12 @@ from user_engine.domain import (
    Catalog,
    ExternalIdentity,
    FamilyInvitation,
+    IdentityFactor,
    Membership,
    OutboxEvent,
    ProfileScope,
    ProfileValue,
+    RegistrationSession,
    TenantAccount,
    User,
 )
@@ -48,6 +50,10 @@ class InMemoryUserEngineStore:
    bindings: dict[str, ApplicationBinding] = field(default_factory=dict)
    catalogs: dict[str, Catalog] = field(default_factory=dict)
    family_invitations: dict[str, FamilyInvitation] = field(default_factory=dict)
+    registration_sessions: dict[str, RegistrationSession] = field(
+        default_factory=dict
+    )
+    identity_factors: dict[str, IdentityFactor] = field(default_factory=dict)
    profile_values: dict[
        tuple[str, str, ProfileScope, str | None], ProfileValue
    ] = field(default_factory=dict)
@@ -142,6 +148,39 @@ class InMemoryUserEngineStore:
            if invitation.user_id == user_id
        )

+    def save_registration_session(self, session: RegistrationSession) -> None:
+        self.registration_sessions[session.registration_id] = session
+
+    def registration_session(
+        self, registration_id: str
+    ) -> RegistrationSession | None:
+        return self.registration_sessions.get(registration_id)
+
+    def all_registration_sessions(self) -> tuple[RegistrationSession, ...]:
+        return tuple(self.registration_sessions.values())
+
+    def save_identity_factor(self, factor: IdentityFactor) -> None:
+        self.identity_factors[factor.factor_id] = factor
+
+    def identity_factor(self, factor_id: str) -> IdentityFactor | None:
+        return self.identity_factors.get(factor_id)
+
+    def factors_for_registration(
+        self, registration_id: str
+    ) -> tuple[IdentityFactor, ...]:
+        return tuple(
+            factor
+            for factor in self.identity_factors.values()
+            if factor.registration_id == registration_id
+        )
+
+    def factors_for_user(self, user_id: str) -> tuple[IdentityFactor, ...]:
+        return tuple(
+            factor
+            for factor in self.identity_factors.values()
+            if factor.user_id == user_id
+        )
+
    def save_profile_value(self, value: ProfileValue) -> None:
        self.profile_values[
            (value.user_id, value.attribute_key, value.scope, value.scope_id)
@@ -199,6 +238,8 @@ class InMemoryUserEngineStore:
            "applications": len(self.applications),
            "catalogs": len(self.catalogs),
            "family_invitations": len(self.family_invitations),
+            "registration_sessions": len(self.registration_sessions),
+            "identity_factors": len(self.identity_factors),
            "profile_values": len(self.profile_values),
            "audit_records": len(self.audit_records),
            "pending_outbox_events": len(self.outbox_events),
@@ -215,6 +256,8 @@ class InMemoryUserEngineStore:
            "bindings": copy.deepcopy(self.bindings),
            "catalogs": copy.deepcopy(self.catalogs),
            "family_invitations": copy.deepcopy(self.family_invitations),
+            "registration_sessions": copy.deepcopy(self.registration_sessions),
+            "identity_factors": copy.deepcopy(self.identity_factors),
            "profile_values": copy.deepcopy(self.profile_values),
            "audit_records": copy.deepcopy(self.audit_records),
            "outbox_events": copy.deepcopy(self.outbox_events),
@@ -238,6 +281,10 @@ class InMemoryUserEngineStore:
        self.family_invitations = snapshot[
            "family_invitations"
        ]  # type: ignore[assignment]
+        self.registration_sessions = snapshot[
+            "registration_sessions"
+        ]  # type: ignore[assignment]
+        self.identity_factors = snapshot["identity_factors"]  # type: ignore[assignment]
        self.profile_values = snapshot["profile_values"]  # type: ignore[assignment]
        self.audit_records = [*snapshot_audit_records, *denied_audit_records]
        self.outbox_events = snapshot["outbox_events"]  # type: ignore[assignment]
--- a/src/user_engine/domain/init.py
+++ b/src/user_engine/domain/init.py
@@ -16,10 +16,13 @@ from user_engine.domain.models import (
    Catalog,
    CatalogLifecycle,
    ExternalIdentity,
+    FactorVerification,
    FamilyDataspaceRequest,
    FamilyInvitation,
    FamilyMemberSpec,
    FamilyRole,
+    IdentityFactor,
+    IdentityFactorType,
    InvitationStatus,
    ManagementMode,
    Membership,
@@ -29,6 +32,8 @@ from user_engine.domain.models import (
    ProfileScope,
    ProfileValue,
    ProjectionType,
+    RegistrationSession,
+    RegistrationStatus,
    Sensitivity,
    TenantAccount,
    User,
@@ -53,10 +58,13 @@ __all__ = [
    "Catalog",
    "CatalogLifecycle",
    "ExternalIdentity",
+    "FactorVerification",
    "FamilyDataspaceRequest",
    "FamilyInvitation",
    "FamilyMemberSpec",
    "FamilyRole",
+    "IdentityFactor",
+    "IdentityFactorType",
    "InvitationStatus",
    "ManagementMode",
    "Membership",
@@ -66,6 +74,8 @@ __all__ = [
    "ProfileScope",
    "ProfileValue",
    "ProjectionType",
+    "RegistrationSession",
+    "RegistrationStatus",
    "Sensitivity",
    "TenantAccount",
    "User",
--- a/src/user_engine/domain/models.py
+++ b/src/user_engine/domain/models.py
@@ -59,6 +59,25 @@ class InvitationStatus(StrEnum):
    REVOKED = "revoked"


+class RegistrationStatus(StrEnum):
+    STARTED = "started"
+    FACTOR_PENDING = "factor_pending"
+    FACTOR_VERIFIED = "factor_verified"
+    COMPLETED = "completed"
+    ABANDONED = "abandoned"
+    EXPIRED = "expired"
+    REJECTED = "rejected"
+
+
+class IdentityFactorType(StrEnum):
+    EMAIL = "email"
+    PHONE = "phone"
+    POSTAL_ADDRESS = "postal_address"
+    EID = "eid"
+    INVITE = "invite"
+    SSO = "sso"
+
+
 class ProfileScope(StrEnum):
    GLOBAL = "global"
    TENANT = "tenant"
@@ -313,6 +332,53 @@ class FamilyInvitation:
    revoked_at: datetime | None = None


+@dataclass(frozen=True)
+class FactorVerification:
+    factor_type: IdentityFactorType
+    normalized_value: str
+    verification_id: str = field(default_factory=lambda: new_id("fvr"))
+    display_value: str | None = None
+    source_system: str = "external-proofing"
+    assurance: Mapping[str, Any] = field(default_factory=dict)
+    evidence_refs: tuple[CanonEntityReference, ...] = ()
+    verified_at: datetime = field(default_factory=utc_now)
+    expires_at: datetime | None = None
+
+
+@dataclass(frozen=True)
+class IdentityFactor:
+    factor_type: IdentityFactorType
+    normalized_value: str
+    factor_id: str = field(default_factory=lambda: new_id("fac"))
+    registration_id: str | None = None
+    user_id: str | None = None
+    display_value: str | None = None
+    source_system: str = "external-proofing"
+    assurance: Mapping[str, Any] = field(default_factory=dict)
+    evidence_refs: tuple[CanonEntityReference, ...] = ()
+    verified_at: datetime = field(default_factory=utc_now)
+    expires_at: datetime | None = None
+
+
+@dataclass(frozen=True)
+class RegistrationSession:
+    tenant: str
+    registration_id: str = field(default_factory=lambda: new_id("reg"))
+    status: RegistrationStatus = RegistrationStatus.STARTED
+    required_factor_types: tuple[IdentityFactorType, ...] = ()
+    verified_factor_ids: tuple[str, ...] = ()
+    user_id: str | None = None
+    netkingdom_id: str | None = None
+    started_by_subject: str | None = None
+    correlation_id: str | None = None
+    created_at: datetime = field(default_factory=utc_now)
+    updated_at: datetime = field(default_factory=utc_now)
+    completed_at: datetime | None = None
+    abandoned_at: datetime | None = None
+    expired_at: datetime | None = None
+    rejected_at: datetime | None = None
+
+
@dataclass(frozen=True)
 class AuthorizationRequest:
    actor: Actor
--- a/src/user_engine/ports.py
+++ b/src/user_engine/ports.py
@@ -21,10 +21,13 @@ from user_engine.domain import (
    CanonEntityReference,
    Catalog,
    ExternalIdentity,
+    FactorVerification,
    FamilyInvitation,
+    IdentityFactor,
    Membership,
    OutboxEvent,
    ProfileValue,
+    RegistrationSession,
    TenantAccount,
    User,
 )
@@ -120,6 +123,31 @@ class UserEngineStore(Protocol):
    ) -> tuple[FamilyInvitation, ...]:
        """Return family invitations for a user."""

+    def save_registration_session(self, session: RegistrationSession) -> None:
+        """Create or replace a registration session."""
+
+    def registration_session(
+        self, registration_id: str
+    ) -> RegistrationSession | None:
+        """Return a registration session by id."""
+
+    def all_registration_sessions(self) -> tuple[RegistrationSession, ...]:
+        """Return all registration sessions."""
+
+    def save_identity_factor(self, factor: IdentityFactor) -> None:
+        """Create or replace a verified identity factor."""
+
+    def identity_factor(self, factor_id: str) -> IdentityFactor | None:
+        """Return a verified identity factor by id."""
+
+    def factors_for_registration(
+        self, registration_id: str
+    ) -> tuple[IdentityFactor, ...]:
+        """Return verified factors attached to a registration session."""
+
+    def factors_for_user(self, user_id: str) -> tuple[IdentityFactor, ...]:
+        """Return verified factors attached to a user."""
+
    def save_profile_value(self, value: ProfileValue) -> None:
        """Create or replace a profile value."""

@@ -152,6 +180,13 @@ class IdentityClaimsAdapter(Protocol):
        """Return the stable external identity link key."""


+class FactorVerificationAdapter(Protocol):
+    """Normalize external proofing results into safe factor evidence."""
+
+    def normalize(self, proofing_result: Mapping[str, Any]) -> FactorVerification:
+        """Return normalized verified factor evidence without secret payloads."""
+
+
 class AuthorizationCheckPort(Protocol):
    """Ask whether an actor may perform an action."""

--- a/src/user_engine/service.py
+++ b/src/user_engine/service.py
@@ -3,7 +3,7 @@
 from __future__ import annotations

 from dataclasses import dataclass, replace
-from typing import Any, Mapping
+from typing import Any, Iterable, Mapping

 from user_engine.domain import (
    Account,
@@ -19,10 +19,13 @@ from user_engine.domain import (
    Catalog,
    CatalogLifecycle,
    ExternalIdentity,
+    FactorVerification,
    FamilyDataspaceRequest,
    FamilyInvitation,
    FamilyMemberSpec,
    FamilyRole,
+    IdentityFactor,
+    IdentityFactorType,
    InvitationStatus,
    Membership,
    Mutability,
@@ -30,6 +33,8 @@ from user_engine.domain import (
    ProfileScope,
    ProfileValue,
    ProjectionType,
+    RegistrationSession,
+    RegistrationStatus,
    Sensitivity,
    TenantAccount,
    User,
@@ -45,6 +50,7 @@ from user_engine.errors import (
 )
 from user_engine.ports import (
    AuthorizationCheckPort,
+    FactorVerificationAdapter,
    IdentityClaimsAdapter,
    UserEngineStore,
 )
@@ -53,6 +59,12 @@ REDACTED = "<redacted>"
 PLATFORM_TENANT = "platform:root"
 PLATFORM_OPERATOR_ROLE = "platform-operator"
 TENANT_ADMIN_ROLE = "tenant-admin"
+_TERMINAL_REGISTRATION_STATUSES = {
+    RegistrationStatus.COMPLETED,
+    RegistrationStatus.ABANDONED,
+    RegistrationStatus.EXPIRED,
+    RegistrationStatus.REJECTED,
+}


@dataclass(frozen=True)
@@ -148,6 +160,15 @@ class FamilyInvitationAcceptance:
    claims_projection: Projection


+@dataclass(frozen=True)
+class RegistrationCompletion:
+    session: RegistrationSession
+    user: User
+    account: Account
+    netkingdom_id: str
+    identity_context: IdentityContext
+
+
@dataclass(frozen=True)
 class TenantDiagnostics:
    tenant: str
@@ -163,6 +184,15 @@ class OutboxDiagnostics:
    oldest_correlation_id: str | None


+@dataclass(frozen=True)
+class RegistrationDiagnostics:
+    tenant: str
+    total_sessions: int
+    statuses: Mapping[str, int]
+    verified_factor_count: int
+    pending_session_count: int
+
+
@dataclass(frozen=True)
 class OperabilitySnapshot:
    ready: bool
@@ -180,10 +210,12 @@ class UserEngineService:
        store: UserEngineStore,
        identity_adapter: IdentityClaimsAdapter,
        authorization: AuthorizationCheckPort,
+        factor_verifier: FactorVerificationAdapter | None = None,
    ) -> None:
        self.store = store
        self.identity_adapter = identity_adapter
        self.authorization = authorization
+        self.factor_verifier = factor_verifier
        if not self.store.ready:
            self.store.migrate()

@@ -217,6 +249,369 @@ class UserEngineService:
            platform_operator=platform_operator,
        )

+    def start_registration(
+        self,
+        actor: Actor,
+        *,
+        tenant: str | None = None,
+        required_factor_types: Iterable[IdentityFactorType | str] = (
+            IdentityFactorType.EMAIL,
+        ),
+        correlation_id: str | None = None,
+    ) -> RegistrationSession:
+        tenant_context = self.resolve_tenant_context(actor, tenant)
+        correlation_id = correlation_id or new_id("corr")
+        required = tuple(_factor_type(item) for item in required_factor_types)
+        decision = self._authorize(
+            actor,
+            action="registration.start",
+            resource_type="user-engine:registration",
+            resource_id="new",
+            tenant=tenant_context.tenant,
+            correlation_id=correlation_id,
+            context={"required_factor_types": tuple(item.value for item in required)},
+        )
+        session = RegistrationSession(
+            tenant=tenant_context.tenant,
+            status=(
+                RegistrationStatus.FACTOR_PENDING
+                if required
+                else RegistrationStatus.STARTED
+            ),
+            required_factor_types=required,
+            started_by_subject=actor.subject,
+            correlation_id=correlation_id,
+        )
+        with self.store.transaction():
+            self.store.save_registration_session(session)
+            self._record_mutation(
+                actor,
+                action="registration.start",
+                subject=session.registration_id,
+                tenant=tenant_context.tenant,
+                correlation_id=correlation_id,
+                decision_id=decision.decision_id,
+                event_type="registration.started",
+                aggregate_id=session.registration_id,
+                payload={
+                    "registration_id": session.registration_id,
+                    "status": session.status,
+                    "required_factor_types": tuple(
+                        item.value for item in session.required_factor_types
+                    ),
+                },
+            )
+        return session
+
+    def attach_registration_factor(
+        self,
+        actor: Actor,
+        registration_id: str,
+        verification: FactorVerification | Mapping[str, Any],
+        *,
+        correlation_id: str | None = None,
+    ) -> RegistrationSession:
+        session = self._require_registration_session(registration_id)
+        self._ensure_registration_mutable(session)
+        tenant_context = self.resolve_tenant_context(actor, session.tenant)
+        correlation_id = correlation_id or new_id("corr")
+        verified = self._normalize_factor_verification(verification)
+        if not verified.normalized_value:
+            raise ValidationError("verified factor normalized_value is required")
+        decision = self._authorize(
+            actor,
+            action="registration.factor.write",
+            resource_type="user-engine:registration-factor",
+            resource_id=f"{registration_id}:{verified.factor_type.value}",
+            tenant=tenant_context.tenant,
+            correlation_id=correlation_id,
+            context={
+                "registration_id": registration_id,
+                "factor_type": verified.factor_type.value,
+                "source_system": verified.source_system,
+            },
+        )
+        factor = IdentityFactor(
+            factor_type=verified.factor_type,
+            normalized_value=verified.normalized_value,
+            registration_id=session.registration_id,
+            display_value=verified.display_value,
+            source_system=verified.source_system,
+            assurance=dict(verified.assurance),
+            evidence_refs=verified.evidence_refs,
+            verified_at=verified.verified_at,
+            expires_at=verified.expires_at,
+        )
+        current_factor_ids = tuple(
+            factor.factor_id for factor in self.store.factors_for_registration(
+                session.registration_id
+            )
+        )
+        verified_factor_ids = (*current_factor_ids, factor.factor_id)
+        all_factor_types = {
+            item.factor_type
+            for item in self.store.factors_for_registration(session.registration_id)
+        }
+        all_factor_types.add(factor.factor_type)
+        next_status = (
+            RegistrationStatus.FACTOR_VERIFIED
+            if set(session.required_factor_types).issubset(all_factor_types)
+            else RegistrationStatus.FACTOR_PENDING
+        )
+        updated = replace(
+            session,
+            status=next_status,
+            verified_factor_ids=verified_factor_ids,
+            updated_at=utc_now(),
+        )
+        with self.store.transaction():
+            self.store.save_identity_factor(factor)
+            self.store.save_registration_session(updated)
+            self._record_mutation(
+                actor,
+                action="registration.factor.write",
+                subject=session.registration_id,
+                tenant=tenant_context.tenant,
+                correlation_id=correlation_id,
+                decision_id=decision.decision_id,
+                event_type="registration.factor_verified",
+                aggregate_id=session.registration_id,
+                payload={
+                    "registration_id": session.registration_id,
+                    "factor_id": factor.factor_id,
+                    "factor_type": factor.factor_type,
+                    "source_system": factor.source_system,
+                    "evidence_ref_count": len(factor.evidence_refs),
+                    "status": updated.status,
+                },
+            )
+        return updated
+
+    def complete_registration(
+        self,
+        actor: Actor,
+        registration_id: str,
+        *,
+        display_name: str | None = None,
+        primary_email: str | None = None,
+        correlation_id: str | None = None,
+    ) -> RegistrationCompletion:
+        session = self._require_registration_session(registration_id)
+        self._ensure_registration_mutable(session)
+        tenant_context = self.resolve_tenant_context(actor, session.tenant)
+        factors = self.store.factors_for_registration(session.registration_id)
+        missing = _missing_required_factor_types(session, factors)
+        if missing:
+            raise ValidationError(
+                "registration is missing required factors: "
+                + ", ".join(item.value for item in missing)
+            )
+        correlation_id = correlation_id or new_id("corr")
+        decision = self._authorize(
+            actor,
+            action="registration.complete",
+            resource_type="user-engine:registration",
+            resource_id=session.registration_id,
+            tenant=tenant_context.tenant,
+            correlation_id=correlation_id,
+            context={
+                "factor_types": tuple(
+                    sorted({factor.factor_type.value for factor in factors})
+                )
+            },
+        )
+        existing_identity = self.store.find_identity(*actor.identity_key)
+        if existing_identity is not None:
+            user = self._require_user(existing_identity.user_id)
+            account = self._require_account(user.user_id)
+            tenant_account = self.store.tenant_account(
+                tenant_context.tenant, user.user_id
+            )
+            identity = existing_identity
+        else:
+            user = User(
+                display_name=display_name or actor.preferred_username,
+                primary_email=primary_email or _registration_primary_email(
+                    factors, actor
+                ),
+            )
+            account = Account(
+                account_id=new_id("acct"),
+                user_id=user.user_id,
+                status=AccountStatus.ACTIVE,
+            )
+            tenant_account = TenantAccount(
+                user_id=user.user_id,
+                tenant=tenant_context.tenant,
+                status=AccountStatus.ACTIVE,
+            )
+            identity = ExternalIdentity(
+                identity_id=new_id("idn"),
+                user_id=user.user_id,
+                issuer=actor.issuer,
+                subject=actor.subject,
+                provider=actor.authorized_party,
+            )
+        completed_at = utc_now()
+        completed = replace(
+            session,
+            status=RegistrationStatus.COMPLETED,
+            user_id=user.user_id,
+            netkingdom_id=user.user_id,
+            updated_at=completed_at,
+            completed_at=completed_at,
+        )
+        attached_factors = tuple(
+            replace(factor, user_id=user.user_id) for factor in factors
+        )
+        with self.store.transaction():
+            if existing_identity is None:
+                self.store.save_user(user)
+                self.store.save_account(account)
+                self.store.save_tenant_account(tenant_account)
+                self.store.save_identity(identity)
+            elif tenant_account is None:
+                self.store.save_tenant_account(
+                    TenantAccount(
+                        user_id=user.user_id,
+                        tenant=tenant_context.tenant,
+                        status=AccountStatus.ACTIVE,
+                    )
+                )
+            for factor in attached_factors:
+                self.store.save_identity_factor(factor)
+            self.store.save_registration_session(completed)
+            self._record_mutation(
+                actor,
+                action="registration.complete",
+                subject=session.registration_id,
+                tenant=tenant_context.tenant,
+                correlation_id=correlation_id,
+                decision_id=decision.decision_id,
+                event_type="registration.completed",
+                aggregate_id=user.user_id,
+                payload={
+                    "registration_id": session.registration_id,
+                    "netkingdom_id": user.user_id,
+                    "user_id": user.user_id,
+                    "factor_types": tuple(
+                        sorted({factor.factor_type.value for factor in factors})
+                    ),
+                },
+            )
+        identity_context = self.identity_context(
+            actor,
+            user_id=user.user_id,
+            tenant=tenant_context.tenant,
+            correlation_id=correlation_id,
+        )
+        return RegistrationCompletion(
+            session=completed,
+            user=user,
+            account=account,
+            netkingdom_id=user.user_id,
+            identity_context=identity_context,
+        )
+
+    def abandon_registration(
+        self,
+        actor: Actor,
+        registration_id: str,
+        *,
+        correlation_id: str | None = None,
+    ) -> RegistrationSession:
+        return self._close_registration(
+            actor,
+            registration_id,
+            status=RegistrationStatus.ABANDONED,
+            action="registration.abandon",
+            event_type="registration.abandoned",
+            correlation_id=correlation_id,
+        )
+
+    def expire_registration(
+        self,
+        actor: Actor,
+        registration_id: str,
+        *,
+        correlation_id: str | None = None,
+    ) -> RegistrationSession:
+        return self._close_registration(
+            actor,
+            registration_id,
+            status=RegistrationStatus.EXPIRED,
+            action="registration.expire",
+            event_type="registration.expired",
+            correlation_id=correlation_id,
+        )
+
+    def resume_registration(
+        self,
+        actor: Actor,
+        registration_id: str,
+        *,
+        correlation_id: str | None = None,
+    ) -> RegistrationSession:
+        session = self._require_registration_session(registration_id)
+        if session.status in _TERMINAL_REGISTRATION_STATUSES:
+            raise ValidationError("terminal registration sessions cannot be resumed")
+        tenant_context = self.resolve_tenant_context(actor, session.tenant)
+        correlation_id = correlation_id or new_id("corr")
+        self._authorize(
+            actor,
+            action="registration.resume",
+            resource_type="user-engine:registration",
+            resource_id=session.registration_id,
+            tenant=tenant_context.tenant,
+            correlation_id=correlation_id,
+        )
+        return session
+
+    def registration_diagnostics(
+        self,
+        actor: Actor,
+        *,
+        tenant: str,
+        correlation_id: str | None = None,
+    ) -> RegistrationDiagnostics:
+        tenant_context = self.resolve_tenant_context(actor, tenant)
+        correlation_id = correlation_id or new_id("corr")
+        self._authorize(
+            actor,
+            action="registration.diagnostics.read",
+            resource_type="user-engine:registration",
+            resource_id=tenant_context.tenant,
+            tenant=tenant_context.tenant,
+            correlation_id=correlation_id,
+        )
+        sessions = tuple(
+            session
+            for session in self.store.all_registration_sessions()
+            if session.tenant == tenant_context.tenant
+        )
+        statuses: dict[str, int] = {}
+        verified_factor_count = 0
+        for session in sessions:
+            statuses[session.status.value] = statuses.get(session.status.value, 0) + 1
+            verified_factor_count += len(
+                self.store.factors_for_registration(session.registration_id)
+            )
+        pending_count = sum(
+            statuses.get(status.value, 0)
+            for status in (
+                RegistrationStatus.STARTED,
+                RegistrationStatus.FACTOR_PENDING,
+                RegistrationStatus.FACTOR_VERIFIED,
+            )
+        )
+        return RegistrationDiagnostics(
+            tenant=tenant_context.tenant,
+            total_sessions=len(sessions),
+            statuses=statuses,
+            verified_factor_count=verified_factor_count,
+            pending_session_count=pending_count,
+        )
+
    def me(
        self,
        claims: Mapping[str, Any],
@@ -1305,6 +1700,76 @@ class UserEngineService:
            context["actor_subject"] = actor.subject
        return context

+    def _close_registration(
+        self,
+        actor: Actor,
+        registration_id: str,
+        *,
+        status: RegistrationStatus,
+        action: str,
+        event_type: str,
+        correlation_id: str | None,
+    ) -> RegistrationSession:
+        session = self._require_registration_session(registration_id)
+        self._ensure_registration_mutable(session)
+        tenant_context = self.resolve_tenant_context(actor, session.tenant)
+        correlation_id = correlation_id or new_id("corr")
+        decision = self._authorize(
+            actor,
+            action=action,
+            resource_type="user-engine:registration",
+            resource_id=session.registration_id,
+            tenant=tenant_context.tenant,
+            correlation_id=correlation_id,
+        )
+        now = utc_now()
+        updated = replace(
+            session,
+            status=status,
+            updated_at=now,
+            abandoned_at=now if status == RegistrationStatus.ABANDONED else None,
+            expired_at=now if status == RegistrationStatus.EXPIRED else None,
+            rejected_at=now if status == RegistrationStatus.REJECTED else None,
+        )
+        with self.store.transaction():
+            self.store.save_registration_session(updated)
+            self._record_mutation(
+                actor,
+                action=action,
+                subject=session.registration_id,
+                tenant=tenant_context.tenant,
+                correlation_id=correlation_id,
+                decision_id=decision.decision_id,
+                event_type=event_type,
+                aggregate_id=session.registration_id,
+                payload={
+                    "registration_id": session.registration_id,
+                    "status": updated.status,
+                },
+            )
+        return updated
+
+    def _require_registration_session(
+        self, registration_id: str
+    ) -> RegistrationSession:
+        session = self.store.registration_session(registration_id)
+        if session is None:
+            raise NotFoundError("registration session not found")
+        return session
+
+    def _ensure_registration_mutable(self, session: RegistrationSession) -> None:
+        if session.status in _TERMINAL_REGISTRATION_STATUSES:
+            raise ValidationError("registration session is already terminal")
+
+    def _normalize_factor_verification(
+        self, verification: FactorVerification | Mapping[str, Any]
+    ) -> FactorVerification:
+        if isinstance(verification, FactorVerification):
+            return verification
+        if self.factor_verifier is None:
+            raise ValidationError("factor verifier adapter is required")
+        return self.factor_verifier.normalize(verification)
+
    def _ensure_actor_session(
        self, actor: Actor, correlation_id: str
    ) -> UserSession:
@@ -2103,6 +2568,33 @@ def _optional_claim(actor: Actor, key: str) -> str | None:
    return str(value)


+def _factor_type(value: IdentityFactorType | str) -> IdentityFactorType:
+    try:
+        return IdentityFactorType(str(value))
+    except ValueError as exc:
+        raise ValidationError(f"unsupported identity factor type: {value}") from exc
+
+
+def _missing_required_factor_types(
+    session: RegistrationSession, factors: Iterable[IdentityFactor]
+) -> tuple[IdentityFactorType, ...]:
+    verified = {factor.factor_type for factor in factors}
+    return tuple(
+        factor_type
+        for factor_type in session.required_factor_types
+        if factor_type not in verified
+    )
+
+
+def _registration_primary_email(
+    factors: Iterable[IdentityFactor], actor: Actor
+) -> str | None:
+    for factor in factors:
+        if factor.factor_type == IdentityFactorType.EMAIL:
+            return factor.display_value or factor.normalized_value
+    return _optional_claim(actor, "email")
+
+
 def _scope_concept(scope_type: str) -> str:
    concepts = {
        "team": "Team",