user-engine/src/user_engine/service.py

"""Headless application service for the isolated user-engine MVP."""

from __future__ import annotations

from dataclasses import dataclass, field, replace
from datetime import datetime, timedelta
from typing import Any, Callable, Iterable, Mapping

from user_engine.domain import (
    Account,
    AccessControlFact,
    AccessMembershipRequirement,
    AccessProfile,
    AccessScopeType,
    AccountStatus,
    ActiveAccessContext,
    Actor,
    Application,
    ApplicationBinding,
    AttributeDefinition,
    AuditRecord,
    AuthorizationDecision,
    AuthorizationRequest,
    CanonEntityReference,
    CanonRelationshipReference,
    Catalog,
    CatalogLifecycle,
    ExternalIdentity,
    FactorVerification,
    FamilyDataspaceRequest,
    FamilyInvitation,
    FamilyMemberSpec,
    FamilyRole,
    IdentityFactor,
    IdentityFactorType,
    InvitationStatus,
    Membership,
    Mutability,
    OnboardingJourney,
    OnboardingJourneyStatus,
    OnboardingStep,
    OnboardingStepStatus,
    OnboardingTask,
    OnboardingTriggerType,
    OutboxEvent,
    PreparedAccount,
    PreparedAccountStatus,
    PreparedEntitlement,
    PreparedEntitlementKind,
    PreparedFactorRequirement,
    ProfileScope,
    ProfileValue,
    ProjectionType,
    RegistrationSession,
    RegistrationStatus,
    Sensitivity,
    SubsystemHandoff,
    TenantAccount,
    User,
    Visibility,
    WelcomeProtocol,
    WelcomeProtocolStep,
    new_id,
    utc_now,
)
from user_engine.errors import (
    AuthorizationDenied,
    ConflictError,
    NotFoundError,
    ValidationError,
)
from user_engine.evidence import (
    HEARTBEAT_EVENT_TYPE,
    load_bearing_counts,
)
from user_engine.ports import (
    AuthorizationCheckPort,
    FactorVerificationAdapter,
    IdentityClaimsAdapter,
    UserEngineStore,
)

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,
}
_UNSET = object()


@dataclass(frozen=True)
class HealthReport:
    status: str
    schema_version: str | None


@dataclass(frozen=True)
class ReadinessReport:
    ready: bool
    checks: Mapping[str, bool]


@dataclass(frozen=True)
class UserSession:
    actor: Actor
    user: User
    account: Account
    identities: tuple[ExternalIdentity, ...]


@dataclass(frozen=True)
class TenantContext:
    tenant: str
    actor: Actor
    platform_operator: bool


@dataclass(frozen=True)
class EffectiveProfile:
    user_id: str
    application_id: str | None
    values: Mapping[str, Any]
    source_layers: Mapping[str, str]
    trace: Mapping[str, tuple[str, ...]]


@dataclass(frozen=True)
class Projection:
    user_id: str
    projection_type: ProjectionType
    application_id: str | None
    values: Mapping[str, Any]
    redactions: Mapping[str, str]
    access_context: Mapping[str, Any] = field(default_factory=dict)


@dataclass(frozen=True)
class IdentityContext:
    actor: Actor
    user: User
    account: Account
    identities: tuple[ExternalIdentity, ...]
    tenant: str
    application_id: str | None
    memberships: tuple[Membership, ...]
    entity_refs: Mapping[str, CanonEntityReference]
    relationship_refs: tuple[CanonRelationshipReference, ...]
    grant_like_refs: tuple[CanonEntityReference, ...]
    evidence_refs: tuple[CanonEntityReference, ...]
    profile: EffectiveProfile | None = None
    gaps: tuple[str, ...] = ()
    active_access_context: ActiveAccessContext | None = None
    access_control_facts: tuple[AccessControlFact, ...] = ()
    onboarding_journeys: tuple[OnboardingJourney, ...] = ()


@dataclass(frozen=True)
class AccessProfileSelection:
    profile: AccessProfile
    active_context: ActiveAccessContext
    access_control_facts: tuple[AccessControlFact, ...]


@dataclass(frozen=True)
class AccessControlFactExport:
    tenant: str
    facts: tuple[AccessControlFact, ...]
    manifest: Mapping[str, Any]


@dataclass(frozen=True)
class AccessProfileDiagnostics:
    tenant: str
    profile_count: int
    active_context_count: int
    required_factor_types: Mapping[str, tuple[str, ...]]
    issues: tuple[str, ...]


@dataclass(frozen=True)
class OnboardingJourneyStart:
    journey: OnboardingJourney
    protocol: WelcomeProtocol


@dataclass(frozen=True)
class OnboardingDiagnostics:
    tenant: str
    protocol_count: int
    journey_count: int
    statuses: Mapping[str, int]
    blocked_steps: tuple[str, ...]
    lifecycle_gaps: tuple[str, ...]


@dataclass(frozen=True)
class FamilyMemberInvitation:
    user: User
    tenant_account: TenantAccount
    membership: Membership
    invitation: FamilyInvitation


@dataclass(frozen=True)
class FamilyDataspaceOnboarding:
    tenant: str
    family_scope_id: str
    family_display_name: str
    owner_session: UserSession
    owner_membership: Membership
    application: Application
    binding: ApplicationBinding
    catalog: Catalog
    invitations: tuple[FamilyMemberInvitation, ...]
    identity_context: IdentityContext
    claims_projection: Projection


@dataclass(frozen=True)
class FamilyInvitationAcceptance:
    session: UserSession
    invitation: FamilyInvitation
    identity_context: IdentityContext
    claims_projection: Projection


@dataclass(frozen=True)
class RegistrationCompletion:
    session: RegistrationSession
    user: User
    account: Account
    netkingdom_id: str
    identity_context: IdentityContext


@dataclass(frozen=True)
class PreparedAccountClaim:
    prepared_account: PreparedAccount
    registration: RegistrationSession
    user: User
    memberships: tuple[Membership, ...]
    tenant_accounts: tuple[TenantAccount, ...]
    profile_values: tuple[ProfileValue, ...]
    application_bindings: tuple[ApplicationBinding, ...]
    onboarding_journeys: tuple[str, ...]


@dataclass(frozen=True)
class TenantDiagnostics:
    tenant: str
    checks: Mapping[str, bool]
    issues: tuple[str, ...]
    memberships: tuple[Membership, ...]


@dataclass(frozen=True)
class OutboxDiagnostics:
    pending_count: int
    event_types: Mapping[str, int]
    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
    checks: Mapping[str, bool]
    metrics: Mapping[str, int]
    issues: tuple[str, ...]


class UserEngineService:
    """Headless service API for isolated user and profile management."""

    def __init__(
        self,
        *,
        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()

    def health(self) -> HealthReport:
        return HealthReport(status="ok", schema_version=self.store.schema_version)

    def readiness(self) -> ReadinessReport:
        return ReadinessReport(
            ready=self.store.ready,
            checks={
                "store_schema": self.store.ready,
                "identity_adapter": self.identity_adapter is not None,
                "authorization_port": self.authorization is not None,
            },
        )

    def resolve_tenant_context(
        self, actor: Actor, tenant: str | None = None
    ) -> TenantContext:
        resolved_tenant = tenant or actor.tenant
        if not resolved_tenant:
            raise ValidationError("tenant context is required")
        platform_operator = PLATFORM_OPERATOR_ROLE in actor.roles
        if resolved_tenant == PLATFORM_TENANT and not platform_operator:
            raise AuthorizationDenied("platform tenant requires platform operator role")
        if resolved_tenant != actor.tenant and not platform_operator:
            raise AuthorizationDenied("cross-tenant access denied")
        return TenantContext(
            tenant=resolved_tenant,
            actor=actor,
            platform_operator=platform_operator,
        )

    def update_self_service_profile(
        self,
        actor: Actor,
        *,
        display_name: str,
        consent_accepted: bool,
        consent_version: str,
        correlation_id: str | None = None,
    ) -> User:
        """Update safe self-service fields; verified email remains immutable."""
        display_name = display_name.strip()
        consent_version = consent_version.strip()
        if not 1 <= len(display_name) <= 200:
            raise ValidationError("display_name must contain 1 to 200 characters")
        if consent_accepted and not consent_version:
            raise ValidationError("consent_version is required when consent is accepted")
        correlation_id = correlation_id or new_id("corr")
        identity = self.store.find_identity(actor.issuer, actor.subject)
        if identity is None:
            raise NotFoundError("current identity is not linked")
        user = self._require_user(identity.user_id)
        decision = self._authorize(
            actor, action="profile.self.update", resource_type="user-engine:user",
            resource_id=user.user_id, tenant=actor.tenant,
            correlation_id=correlation_id, target_user_id=user.user_id,
        )
        now = utc_now()
        updated = replace(
            user, display_name=display_name, updated_at=now,
            profile_completed_at=now,
            consented_at=now if consent_accepted else None,
            consent_version=consent_version if consent_accepted else None,
        )
        with self.store.transaction():
            self.store.save_user(updated)
            self._record_mutation(
                actor, action="profile.self.update", subject=user.user_id,
                tenant=actor.tenant, correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="user.self_service_profile_updated",
                aggregate_id=user.user_id,
                payload={
                    "user_id": user.user_id,
                    "profile_complete": True,
                    "consent_accepted": consent_accepted,
                    "consent_version": consent_version if consent_accepted else None,
                },
            )
        return updated

    def start_registration(
        self,
        actor: Actor,
        *,
        tenant: str | None = None,
        required_factor_types: Iterable[IdentityFactorType | str] = (
            IdentityFactorType.EMAIL,
        ),
        correlation_id: str | None = None,
        applicant_username: str | None = None,
        client_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,
            applicant_username=applicant_username,
            client_id=client_id,
            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})
                    ),
                },
            )
            self._auto_start_onboarding_journeys(
                actor,
                user.user_id,
                self._matching_welcome_protocols_for_registration(completed),
                trigger_type=OnboardingTriggerType.REGISTRATION,
                source_id=session.registration_id,
                source_event_type="registration.completed",
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
            )
        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 prepare_account(
        self,
        actor: Actor,
        *,
        tenant: str,
        required_factor_matches: Iterable[PreparedFactorRequirement],
        entitlements: Iterable[PreparedEntitlement],
        display_name: str | None = None,
        primary_email: str | None = None,
        expires_at: datetime | None = None,
        correlation_id: str | None = None,
    ) -> PreparedAccount:
        tenant_context = self.resolve_tenant_context(actor, tenant)
        required = tuple(required_factor_matches)
        prepared_entitlements = tuple(entitlements)
        self._validate_prepared_factor_requirements(required)
        if not prepared_entitlements:
            raise ValidationError("prepared account requires at least one entitlement")
        self._validate_prepared_entitlements(
            prepared_entitlements, tenant_context.tenant
        )
        self._ensure_no_duplicate_prepared_account(
            tenant_context.tenant,
            required,
            excluding_prepared_account_id=None,
        )
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="prepared_account.create",
            resource_type="user-engine:prepared-account",
            resource_id="new",
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            context={
                "factor_types": tuple(
                    sorted({item.factor_type.value for item in required})
                ),
                "entitlement_count": len(prepared_entitlements),
            },
        )
        prepared = PreparedAccount(
            tenant=tenant_context.tenant,
            required_factor_matches=required,
            entitlements=prepared_entitlements,
            display_name=display_name,
            primary_email=primary_email,
            prepared_by_subject=actor.subject,
            correlation_id=correlation_id,
            expires_at=expires_at,
        )
        with self.store.transaction():
            self.store.save_prepared_account(prepared)
            self._record_mutation(
                actor,
                action="prepared_account.create",
                subject=prepared.prepared_account_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="prepared_account.created",
                aggregate_id=prepared.prepared_account_id,
                payload={
                    "prepared_account_id": prepared.prepared_account_id,
                    "factor_types": tuple(
                        sorted(
                            {
                                item.factor_type.value
                                for item in prepared.required_factor_matches
                            }
                        )
                    ),
                    "entitlement_count": len(prepared.entitlements),
                    "expires_at": prepared.expires_at.isoformat()
                    if prepared.expires_at
                    else None,
                },
            )
        return prepared

    def update_prepared_account(
        self,
        actor: Actor,
        prepared_account_id: str,
        *,
        required_factor_matches: Iterable[PreparedFactorRequirement] | None = None,
        entitlements: Iterable[PreparedEntitlement] | None = None,
        expires_at: datetime | None | object = _UNSET,
        correlation_id: str | None = None,
    ) -> PreparedAccount:
        prepared = self._require_prepared_account(prepared_account_id)
        self._ensure_prepared_account_pending(prepared)
        tenant_context = self.resolve_tenant_context(actor, prepared.tenant)
        required = (
            tuple(required_factor_matches)
            if required_factor_matches is not None
            else prepared.required_factor_matches
        )
        prepared_entitlements = (
            tuple(entitlements) if entitlements is not None else prepared.entitlements
        )
        self._validate_prepared_factor_requirements(required)
        if not prepared_entitlements:
            raise ValidationError("prepared account requires at least one entitlement")
        self._validate_prepared_entitlements(
            prepared_entitlements, tenant_context.tenant
        )
        self._ensure_no_duplicate_prepared_account(
            tenant_context.tenant,
            required,
            excluding_prepared_account_id=prepared.prepared_account_id,
        )
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="prepared_account.update",
            resource_type="user-engine:prepared-account",
            resource_id=prepared.prepared_account_id,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
        )
        updated = replace(
            prepared,
            required_factor_matches=required,
            entitlements=prepared_entitlements,
            expires_at=prepared.expires_at if expires_at is _UNSET else expires_at,
            updated_at=utc_now(),
        )
        with self.store.transaction():
            self.store.save_prepared_account(updated)
            self._record_mutation(
                actor,
                action="prepared_account.update",
                subject=updated.prepared_account_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="prepared_account.updated",
                aggregate_id=updated.prepared_account_id,
                payload={
                    "prepared_account_id": updated.prepared_account_id,
                    "factor_types": tuple(
                        sorted(
                            {
                                item.factor_type.value
                                for item in updated.required_factor_matches
                            }
                        )
                    ),
                    "entitlement_count": len(updated.entitlements),
                    "expires_at": updated.expires_at.isoformat()
                    if updated.expires_at
                    else None,
                },
            )
        return updated

    def list_prepared_accounts(
        self,
        actor: Actor,
        *,
        tenant: str,
        correlation_id: str | None = None,
    ) -> tuple[PreparedAccount, ...]:
        tenant_context = self.resolve_tenant_context(actor, tenant)
        correlation_id = correlation_id or new_id("corr")
        self._authorize(
            actor,
            action="prepared_account.read",
            resource_type="user-engine:prepared-account",
            resource_id=tenant_context.tenant,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
        )
        return self.store.prepared_accounts_for_tenant(tenant_context.tenant)

    def revoke_prepared_account(
        self,
        actor: Actor,
        prepared_account_id: str,
        *,
        correlation_id: str | None = None,
    ) -> PreparedAccount:
        return self._close_prepared_account(
            actor,
            prepared_account_id,
            status=PreparedAccountStatus.REVOKED,
            action="prepared_account.revoke",
            event_type="prepared_account.revoked",
            correlation_id=correlation_id,
        )

    def expire_prepared_account(
        self,
        actor: Actor,
        prepared_account_id: str,
        *,
        correlation_id: str | None = None,
    ) -> PreparedAccount:
        return self._close_prepared_account(
            actor,
            prepared_account_id,
            status=PreparedAccountStatus.EXPIRED,
            action="prepared_account.expire",
            event_type="prepared_account.expired",
            correlation_id=correlation_id,
        )

    def claim_prepared_account(
        self,
        actor: Actor,
        registration_id: str,
        *,
        prepared_account_id: str | None = None,
        correlation_id: str | None = None,
    ) -> PreparedAccountClaim:
        session = self._require_registration_session(registration_id)
        if session.status != RegistrationStatus.COMPLETED or session.user_id is None:
            raise ValidationError("registration must be completed before claim")
        tenant_context = self.resolve_tenant_context(actor, session.tenant)
        correlation_id = correlation_id or new_id("corr")
        factors = self.store.factors_for_registration(session.registration_id)
        prepared = self._resolve_prepared_account_claim(
            actor,
            session,
            factors,
            prepared_account_id=prepared_account_id,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
        )
        if any(entitlement.requires_approval for entitlement in prepared.entitlements):
            self._record_audit_only(
                actor,
                action="prepared_account.claim",
                subject=prepared.prepared_account_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                summary="prepared account claim denied: approval required",
            )
            raise AuthorizationDenied("prepared account claim requires approval")
        decision = self._authorize(
            actor,
            action="prepared_account.claim",
            resource_type="user-engine:prepared-account",
            resource_id=prepared.prepared_account_id,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            target_user_id=session.user_id,
            context={
                "registration_id": session.registration_id,
                "entitlement_count": len(prepared.entitlements),
            },
        )
        user = self._require_user(session.user_id)
        (
            tenant_accounts,
            memberships,
            profile_values,
            application_bindings,
            onboarding_journeys,
        ) = self._prepared_entitlement_records(
            prepared,
            user.user_id,
            correlation_id=correlation_id,
        )
        claimed_at = utc_now()
        claimed = replace(
            prepared,
            status=PreparedAccountStatus.CLAIMED,
            claimed_by_user_id=user.user_id,
            claimed_registration_id=session.registration_id,
            claimed_at=claimed_at,
            updated_at=claimed_at,
        )
        with self.store.transaction():
            for account in tenant_accounts:
                self.store.save_tenant_account(account)
            for membership in memberships:
                self.store.save_membership(membership)
            for value in profile_values:
                self.store.save_profile_value(value)
            for binding in application_bindings:
                self.store.save_binding(binding)
            self.store.save_prepared_account(claimed)
            self._record_mutation(
                actor,
                action="prepared_account.claim",
                subject=prepared.prepared_account_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="prepared_account.claimed",
                aggregate_id=user.user_id,
                payload={
                    "prepared_account_id": prepared.prepared_account_id,
                    "registration_id": session.registration_id,
                    "user_id": user.user_id,
                    "entitlement_count": len(prepared.entitlements),
                },
            )
            for journey in onboarding_journeys:
                self._record_mutation(
                    actor,
                    action="prepared_account.onboarding.start",
                    subject=prepared.prepared_account_id,
                    tenant=tenant_context.tenant,
                    correlation_id=correlation_id,
                    decision_id=decision.decision_id,
                    event_type="prepared_account.onboarding_requested",
                    aggregate_id=user.user_id,
                    payload={
                        "prepared_account_id": prepared.prepared_account_id,
                        "registration_id": session.registration_id,
                        "user_id": user.user_id,
                        "journey": journey,
                    },
                )
            self._auto_start_onboarding_journeys(
                actor,
                user.user_id,
                self._matching_welcome_protocols_for_prepared_account(claimed),
                trigger_type=OnboardingTriggerType.PREPARED_ACCOUNT,
                source_id=prepared.prepared_account_id,
                source_event_type="prepared_account.claimed",
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
            )
        return PreparedAccountClaim(
            prepared_account=claimed,
            registration=session,
            user=user,
            memberships=memberships,
            tenant_accounts=tenant_accounts,
            profile_values=profile_values,
            application_bindings=application_bindings,
            onboarding_journeys=onboarding_journeys,
        )

    def register_access_profile(
        self,
        actor: Actor,
        profile: AccessProfile,
        *,
        correlation_id: str | None = None,
    ) -> AccessProfile:
        tenant_context = self.resolve_tenant_context(actor, profile.tenant)
        self._validate_access_profile(profile, tenant_context.tenant)
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="access_profile.write",
            resource_type="user-engine:access-profile",
            resource_id=profile.access_profile_id,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            context={
                "hat": profile.hat,
                "scope_type": profile.scope_type.value,
                "scope_id": _access_profile_scope_id(profile),
                "required_factor_types": tuple(
                    factor_type.value for factor_type in profile.required_factor_types
                ),
            },
        )
        updated = replace(profile, tenant=tenant_context.tenant, updated_at=utc_now())
        with self.store.transaction():
            self.store.save_access_profile(updated)
            self._record_mutation(
                actor,
                action="access_profile.write",
                subject=updated.access_profile_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="access_profile.registered",
                aggregate_id=updated.access_profile_id,
                payload={
                    "access_profile_id": updated.access_profile_id,
                    "hat": updated.hat,
                    "scope_type": updated.scope_type,
                    "scope_id": _access_profile_scope_id(updated),
                    "required_factor_types": tuple(
                        factor_type.value for factor_type in updated.required_factor_types
                    ),
                    "membership_requirement_count": len(
                        updated.membership_requirements
                    ),
                },
            )
        return updated

    def list_access_profiles(
        self,
        actor: Actor,
        *,
        tenant: str,
        correlation_id: str | None = None,
    ) -> tuple[AccessProfile, ...]:
        tenant_context = self.resolve_tenant_context(actor, tenant)
        correlation_id = correlation_id or new_id("corr")
        self._authorize(
            actor,
            action="access_profile.read",
            resource_type="user-engine:access-profile",
            resource_id=tenant_context.tenant,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
        )
        return self.store.access_profiles_for_tenant(tenant_context.tenant)

    def select_active_hat(
        self,
        actor: Actor,
        user_id: str,
        access_profile_id: str,
        *,
        tenant: str | None = None,
        correlation_id: str | None = None,
    ) -> AccessProfileSelection:
        profile = self._require_access_profile(access_profile_id)
        tenant_context = self.resolve_tenant_context(actor, tenant or profile.tenant)
        if profile.tenant != tenant_context.tenant:
            raise AuthorizationDenied("access profile tenant mismatch")
        if profile.requires_approval:
            raise AuthorizationDenied("access profile requires approval")
        user = self._require_user(user_id)
        account = self._require_account(user.user_id)
        tenant_account = self.store.tenant_account(tenant_context.tenant, user.user_id)
        if tenant_account is None or tenant_account.status != AccountStatus.ACTIVE:
            raise ValidationError("active tenant account is required")
        memberships = self.store.memberships_for_user(
            user.user_id, tenant=tenant_context.tenant
        )
        matched_memberships = self._satisfying_access_memberships(
            profile, memberships
        )
        factor_ids = self._verified_factor_ids_for_user(
            user.user_id, profile.required_factor_types
        )
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="active_access_context.select",
            resource_type="user-engine:active-access-context",
            resource_id=user.user_id,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            target_user_id=user.user_id,
            context={
                "access_profile_id": profile.access_profile_id,
                "hat": profile.hat,
                "scope_type": profile.scope_type.value,
                "scope_id": _access_profile_scope_id(profile),
                "membership_ids": tuple(
                    membership.membership_id for membership in matched_memberships
                ),
                "required_factor_types": tuple(
                    factor_type.value for factor_type in profile.required_factor_types
                ),
            },
        )
        group_ids = _active_group_ids(profile, memberships)
        active_context = ActiveAccessContext(
            user_id=user.user_id,
            tenant=tenant_context.tenant,
            access_profile_id=profile.access_profile_id,
            hat=profile.hat,
            scope_type=profile.scope_type,
            scope_id=_access_profile_scope_id(profile),
            realm_id=profile.realm_id,
            service_id=profile.service_id,
            asset_id=profile.asset_id,
            membership_ids=tuple(
                membership.membership_id for membership in matched_memberships
            ),
            factor_ids=factor_ids,
            group_ids=group_ids,
            claims=dict(profile.claims),
            profile_defaults=dict(profile.profile_defaults),
            selected_by_subject=actor.subject,
        )
        facts = self._access_control_facts(
            tenant_context.tenant,
            user_id=user.user_id,
            memberships=memberships,
            active_context=active_context,
        )
        with self.store.transaction():
            self.store.save_active_access_context(active_context)
            self._record_mutation(
                actor,
                action="active_access_context.select",
                subject=active_context.active_context_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="active_access_context.selected",
                aggregate_id=user.user_id,
                payload={
                    "active_context_id": active_context.active_context_id,
                    "user_id": user.user_id,
                    "access_profile_id": profile.access_profile_id,
                    "hat": profile.hat,
                    "scope_type": profile.scope_type,
                    "scope_id": active_context.scope_id,
                    "membership_ids": active_context.membership_ids,
                    "factor_types": tuple(
                        factor_type.value for factor_type in profile.required_factor_types
                    ),
                },
            )
        return AccessProfileSelection(
            profile=profile,
            active_context=active_context,
            access_control_facts=facts,
        )

    def export_access_control_facts(
        self,
        actor: Actor,
        *,
        tenant: str,
        user_id: str | None = None,
        correlation_id: str | None = None,
    ) -> AccessControlFactExport:
        tenant_context = self.resolve_tenant_context(actor, tenant)
        correlation_id = correlation_id or new_id("corr")
        self._authorize(
            actor,
            action="access_control_facts.export",
            resource_type="user-engine:access-control-facts",
            resource_id=tenant_context.tenant,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            target_user_id=user_id,
        )
        memberships = (
            self.store.memberships_for_user(user_id, tenant=tenant_context.tenant)
            if user_id is not None
            else self.store.memberships_for_tenant(tenant_context.tenant)
        )
        active_context = (
            self.store.active_access_context(user_id, tenant_context.tenant)
            if user_id is not None
            else None
        )
        facts = self._access_control_facts(
            tenant_context.tenant,
            user_id=user_id,
            memberships=memberships,
            active_context=active_context,
        )
        manifest = {
            "tenant": tenant_context.tenant,
            "fact_count": len(facts),
            "scope_types": tuple(sorted({fact.scope_type for fact in facts})),
            "subject_types": tuple(sorted({fact.subject_type for fact in facts})),
        }
        return AccessControlFactExport(
            tenant=tenant_context.tenant,
            facts=facts,
            manifest=manifest,
        )

    def access_profile_diagnostics(
        self,
        actor: Actor,
        *,
        tenant: str,
        correlation_id: str | None = None,
    ) -> AccessProfileDiagnostics:
        tenant_context = self.resolve_tenant_context(actor, tenant)
        correlation_id = correlation_id or new_id("corr")
        self._authorize(
            actor,
            action="access_profile.diagnostics.read",
            resource_type="user-engine:access-profile",
            resource_id=tenant_context.tenant,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
        )
        profiles = self.store.access_profiles_for_tenant(tenant_context.tenant)
        active_contexts = self.store.active_access_contexts_for_tenant(
            tenant_context.tenant
        )
        issues = []
        for profile in profiles:
            if profile.requires_approval:
                issues.append(f"approval-required:{profile.access_profile_id}")
        return AccessProfileDiagnostics(
            tenant=tenant_context.tenant,
            profile_count=len(profiles),
            active_context_count=len(active_contexts),
            required_factor_types={
                profile.access_profile_id: tuple(
                    factor_type.value for factor_type in profile.required_factor_types
                )
                for profile in profiles
            },
            issues=tuple(issues),
        )

    def register_welcome_protocol(
        self,
        actor: Actor,
        protocol: WelcomeProtocol,
        *,
        correlation_id: str | None = None,
    ) -> WelcomeProtocol:
        tenant_context = self.resolve_tenant_context(actor, protocol.tenant)
        self._validate_welcome_protocol(protocol, tenant_context.tenant)
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="welcome_protocol.write",
            resource_type="user-engine:welcome-protocol",
            resource_id=protocol.protocol_id,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            context={
                "trigger_type": protocol.trigger_type.value,
                "journey_key": protocol.journey_key,
                "step_count": len(protocol.steps),
            },
        )
        updated = replace(protocol, tenant=tenant_context.tenant, updated_at=utc_now())
        with self.store.transaction():
            self.store.save_welcome_protocol(updated)
            self._record_mutation(
                actor,
                action="welcome_protocol.write",
                subject=updated.protocol_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="welcome_protocol.registered",
                aggregate_id=updated.protocol_id,
                payload={
                    "protocol_id": updated.protocol_id,
                    "trigger_type": updated.trigger_type,
                    "journey_key": updated.journey_key,
                    "step_count": len(updated.steps),
                },
            )
        return updated

    def list_welcome_protocols(
        self,
        actor: Actor,
        *,
        tenant: str,
        correlation_id: str | None = None,
    ) -> tuple[WelcomeProtocol, ...]:
        tenant_context = self.resolve_tenant_context(actor, tenant)
        correlation_id = correlation_id or new_id("corr")
        self._authorize(
            actor,
            action="welcome_protocol.read",
            resource_type="user-engine:welcome-protocol",
            resource_id=tenant_context.tenant,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
        )
        return self.store.welcome_protocols_for_tenant(tenant_context.tenant)

    def start_onboarding_journey(
        self,
        actor: Actor,
        user_id: str,
        protocol_id: str,
        *,
        trigger_type: OnboardingTriggerType = OnboardingTriggerType.MANUAL,
        source_id: str | None = None,
        source_event_type: str | None = None,
        correlation_id: str | None = None,
    ) -> OnboardingJourneyStart:
        protocol = self._require_welcome_protocol(protocol_id)
        tenant_context = self.resolve_tenant_context(actor, protocol.tenant)
        user = self._require_user(user_id)
        if self._onboarding_journey_exists(user.user_id, protocol.protocol_id, source_id):
            raise ConflictError("onboarding journey already exists for source")
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="onboarding_journey.start",
            resource_type="user-engine:onboarding-journey",
            resource_id=protocol.protocol_id,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            target_user_id=user.user_id,
            context={
                "protocol_id": protocol.protocol_id,
                "trigger_type": trigger_type.value,
                "source_event_type": source_event_type,
            },
        )
        journey = self._build_onboarding_journey(
            protocol,
            user.user_id,
            trigger_type=trigger_type,
            source_id=source_id,
            source_event_type=source_event_type,
            correlation_id=correlation_id,
        )
        with self.store.transaction():
            self.store.save_onboarding_journey(journey)
            self._record_onboarding_journey_started(
                actor,
                journey,
                decision_id=decision.decision_id,
                correlation_id=correlation_id,
            )
        return OnboardingJourneyStart(journey=journey, protocol=protocol)

    def start_onboarding_for_registration(
        self,
        actor: Actor,
        registration_id: str,
        *,
        protocol_id: str | None = None,
        correlation_id: str | None = None,
    ) -> tuple[OnboardingJourneyStart, ...]:
        session = self._require_registration_session(registration_id)
        if session.status != RegistrationStatus.COMPLETED or session.user_id is None:
            raise ValidationError("registration must be completed before onboarding")
        protocols = (
            (self._require_welcome_protocol(protocol_id),)
            if protocol_id is not None
            else self._matching_welcome_protocols_for_registration(session)
        )
        starts = []
        for protocol in protocols:
            starts.append(
                self.start_onboarding_journey(
                    actor,
                    session.user_id,
                    protocol.protocol_id,
                    trigger_type=OnboardingTriggerType.REGISTRATION,
                    source_id=session.registration_id,
                    source_event_type="registration.completed",
                    correlation_id=correlation_id,
                )
            )
        return tuple(starts)

    def start_onboarding_for_prepared_account(
        self,
        actor: Actor,
        prepared_account_id: str,
        *,
        protocol_id: str | None = None,
        correlation_id: str | None = None,
    ) -> tuple[OnboardingJourneyStart, ...]:
        prepared = self._require_prepared_account(prepared_account_id)
        if (
            prepared.status != PreparedAccountStatus.CLAIMED
            or prepared.claimed_by_user_id is None
        ):
            raise ValidationError("prepared account must be claimed before onboarding")
        protocols = (
            (self._require_welcome_protocol(protocol_id),)
            if protocol_id is not None
            else self._matching_welcome_protocols_for_prepared_account(prepared)
        )
        starts = []
        for protocol in protocols:
            starts.append(
                self.start_onboarding_journey(
                    actor,
                    prepared.claimed_by_user_id,
                    protocol.protocol_id,
                    trigger_type=OnboardingTriggerType.PREPARED_ACCOUNT,
                    source_id=prepared.prepared_account_id,
                    source_event_type="prepared_account.claimed",
                    correlation_id=correlation_id,
                )
            )
        return tuple(starts)

    def progress_onboarding_step(
        self,
        actor: Actor,
        journey_id: str,
        step_key: str,
        *,
        correlation_id: str | None = None,
    ) -> OnboardingJourney:
        return self._transition_onboarding_step(
            actor,
            journey_id,
            step_key,
            status=OnboardingStepStatus.IN_PROGRESS,
            action="onboarding_step.progress",
            event_type="onboarding_step.progressed",
            correlation_id=correlation_id,
        )

    def complete_onboarding_step(
        self,
        actor: Actor,
        journey_id: str,
        step_key: str,
        *,
        correlation_id: str | None = None,
    ) -> OnboardingJourney:
        return self._transition_onboarding_step(
            actor,
            journey_id,
            step_key,
            status=OnboardingStepStatus.COMPLETED,
            action="onboarding_step.complete",
            event_type="onboarding_step.completed",
            correlation_id=correlation_id,
        )

    def skip_onboarding_step(
        self,
        actor: Actor,
        journey_id: str,
        step_key: str,
        *,
        correlation_id: str | None = None,
    ) -> OnboardingJourney:
        return self._transition_onboarding_step(
            actor,
            journey_id,
            step_key,
            status=OnboardingStepStatus.SKIPPED,
            action="onboarding_step.skip",
            event_type="onboarding_step.skipped",
            correlation_id=correlation_id,
        )

    def fail_onboarding_step(
        self,
        actor: Actor,
        journey_id: str,
        step_key: str,
        *,
        lifecycle_gap: str | None = None,
        correlation_id: str | None = None,
    ) -> OnboardingJourney:
        return self._transition_onboarding_step(
            actor,
            journey_id,
            step_key,
            status=OnboardingStepStatus.FAILED,
            action="onboarding_step.fail",
            event_type="onboarding_step.failed",
            lifecycle_gap=lifecycle_gap,
            correlation_id=correlation_id,
        )

    def resume_onboarding_journey(
        self,
        actor: Actor,
        journey_id: str,
        *,
        callback_refs: Mapping[str, str] | None = None,
        correlation_id: str | None = None,
    ) -> OnboardingJourney:
        journey = self._require_onboarding_journey(journey_id)
        tenant_context = self.resolve_tenant_context(actor, journey.tenant)
        if journey.status not in (
            OnboardingJourneyStatus.BLOCKED,
            OnboardingJourneyStatus.FAILED,
            OnboardingJourneyStatus.PENDING,
        ):
            raise ValidationError("only pending, blocked, or failed journeys can resume")
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="onboarding_journey.resume",
            resource_type="user-engine:onboarding-journey",
            resource_id=journey.journey_id,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            target_user_id=journey.user_id,
        )
        updated_steps = _resume_onboarding_steps(journey.steps, callback_refs or {})
        updated = replace(
            journey,
            steps=updated_steps,
            status=_journey_status_from_steps(updated_steps),
            active_step_key=_active_onboarding_step_key(updated_steps),
            updated_at=utc_now(),
            failed_at=None,
        )
        with self.store.transaction():
            self.store.save_onboarding_journey(updated)
            self._record_mutation(
                actor,
                action="onboarding_journey.resume",
                subject=updated.journey_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="onboarding_journey.resumed",
                aggregate_id=updated.user_id,
                payload={
                    "journey_id": updated.journey_id,
                    "user_id": updated.user_id,
                    "status": updated.status,
                    "active_step_key": updated.active_step_key,
                },
            )
        return updated

    def onboarding_diagnostics(
        self,
        actor: Actor,
        *,
        tenant: str,
        correlation_id: str | None = None,
    ) -> OnboardingDiagnostics:
        tenant_context = self.resolve_tenant_context(actor, tenant)
        correlation_id = correlation_id or new_id("corr")
        self._authorize(
            actor,
            action="onboarding.diagnostics.read",
            resource_type="user-engine:onboarding",
            resource_id=tenant_context.tenant,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
        )
        protocols = self.store.welcome_protocols_for_tenant(tenant_context.tenant)
        journeys = self.store.onboarding_journeys_for_tenant(tenant_context.tenant)
        statuses: dict[str, int] = {}
        blocked_steps = []
        lifecycle_gaps = []
        for journey in journeys:
            statuses[journey.status.value] = statuses.get(journey.status.value, 0) + 1
            for step in journey.steps:
                if step.status == OnboardingStepStatus.BLOCKED:
                    blocked_steps.append(f"{journey.journey_id}:{step.step_key}")
                if step.lifecycle_gap:
                    lifecycle_gaps.append(step.lifecycle_gap)
        return OnboardingDiagnostics(
            tenant=tenant_context.tenant,
            protocol_count=len(protocols),
            journey_count=len(journeys),
            statuses=statuses,
            blocked_steps=tuple(blocked_steps),
            lifecycle_gaps=tuple(sorted(set(lifecycle_gaps))),
        )

    def me(
        self,
        claims: Mapping[str, Any],
        *,
        correlation_id: str | None = None,
    ) -> UserSession:
        actor = self.identity_adapter.normalize(claims)
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="me.read",
            resource_type="user-engine:me",
            resource_id=actor.subject,
            tenant=actor.tenant,
            correlation_id=correlation_id,
            context={"self": True},
        )
        identity = self.store.find_identity(*actor.identity_key)
        if identity is not None:
            user = self._require_user(identity.user_id)
            account = self._require_account(user.user_id)
            return self._session(actor, user, account)

        user = User(
            display_name=actor.preferred_username,
            primary_email=_optional_claim(actor, "email"),
        )
        account = Account(
            account_id=new_id("acct"),
            user_id=user.user_id,
            status=AccountStatus.ACTIVE,
        )
        tenant_account = TenantAccount(user_id=user.user_id, tenant=actor.tenant)
        identity = ExternalIdentity(
            identity_id=new_id("idn"),
            user_id=user.user_id,
            issuer=actor.issuer,
            subject=actor.subject,
            provider=actor.authorized_party,
        )
        with self.store.transaction():
            self.store.save_user(user)
            self.store.save_account(account)
            self.store.save_tenant_account(tenant_account)
            self.store.save_identity(identity)
            self._record_mutation(
                actor,
                action="user.create_from_identity",
                subject=user.user_id,
                tenant=actor.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="user.created",
                aggregate_id=user.user_id,
                payload={
                    "user_id": user.user_id,
                    "account_id": account.account_id,
                    "identity": {"issuer": actor.issuer, "subject": actor.subject},
                },
            )
        return self._session(actor, user, account)

    def create_user(
        self,
        actor: Actor,
        *,
        display_name: str | None,
        primary_email: str | None,
        correlation_id: str | None = None,
    ) -> User:
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="user.create",
            resource_type="user-engine:user",
            resource_id="new",
            tenant=actor.tenant,
            correlation_id=correlation_id,
        )
        user = User(display_name=display_name, primary_email=primary_email)
        account = Account(account_id=new_id("acct"), user_id=user.user_id)
        tenant_account = TenantAccount(user_id=user.user_id, tenant=actor.tenant)
        with self.store.transaction():
            self.store.save_user(user)
            self.store.save_account(account)
            self.store.save_tenant_account(tenant_account)
            self._record_mutation(
                actor,
                action="user.create",
                subject=user.user_id,
                tenant=actor.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="user.created",
                aggregate_id=user.user_id,
                payload={"user_id": user.user_id, "account_id": account.account_id},
            )
        return user

    def set_account_status(
        self,
        actor: Actor,
        user_id: str,
        status: AccountStatus,
        *,
        correlation_id: str | None = None,
    ) -> Account:
        account = self._require_account(user_id)
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="account.update",
            resource_type="user-engine:account",
            resource_id=account.account_id,
            tenant=actor.tenant,
            correlation_id=correlation_id,
            target_user_id=user_id,
        )
        updated = replace(account, status=status)
        with self.store.transaction():
            self.store.save_account(updated)
            self._record_mutation(
                actor,
                action="account.update",
                subject=user_id,
                tenant=actor.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="account.status_changed",
                aggregate_id=account.account_id,
                payload={"user_id": user_id, "status": status},
            )
        return updated

    def authorize_tenant_member_action(
        self, actor: Actor, user_id: str, *, tenant: str, correlation_id: str,
    ) -> User:
        """Authorize before any credential-provider side effect or target readout."""
        self.resolve_tenant_context(actor, tenant)
        if not {"tenant-admin", PLATFORM_OPERATOR_ROLE}.intersection(actor.roles):
            raise AuthorizationDenied("tenant administrator role required")
        memberships = self.store.memberships_for_user(user_id, tenant=tenant)
        if not any(m.scope_type == "tenant" and m.scope_id == tenant for m in memberships):
            raise NotFoundError("account is not a member of this tenant")
        self._authorize(actor, action="tenant.account.update",
                        resource_type="user-engine:tenant-account",
                        resource_id=f"{tenant}:{user_id}", tenant=tenant,
                        correlation_id=correlation_id, target_user_id=user_id)
        return self._require_user(user_id)

    def validate_tenant_role_change(self, actor: Actor, user_id: str, role: str, *, tenant: str, correlation_id: str) -> None:
        self.authorize_tenant_member_action(actor, user_id, tenant=tenant, correlation_id=correlation_id)
        if role not in {"user", "tenant-admin"}:
            raise ValidationError("Choose User or Tenant administrator.")
        self._authorize(actor, action="membership.write", resource_type="user-engine:membership",
                        resource_id=f"{tenant}:{user_id}:tenant:{tenant}", tenant=tenant,
                        correlation_id=correlation_id, target_user_id=user_id,
                        context={"scope_type":"tenant", "scope_id":tenant, "kind":role})
        if role != "tenant-admin":
            self.require_admin_successor(user_id, tenant=tenant)

    def set_tenant_role(self, actor: Actor, user_id: str, role: str, *, tenant: str, correlation_id: str) -> None:
        with self.store.tenant_lifecycle_guard(tenant):
            self._set_tenant_role(actor, user_id, role, tenant=tenant, correlation_id=correlation_id)

    def _set_tenant_role(self, actor: Actor, user_id: str, role: str, *, tenant: str, correlation_id: str) -> None:
        self.validate_tenant_role_change(actor, user_id, role, tenant=tenant, correlation_id=correlation_id)
        with self.store.transaction():
            for membership in self.store.memberships_for_user(user_id, tenant=tenant):
                if membership.scope_type == "tenant" and membership.scope_id == tenant:
                    self.store.save_membership(replace(membership, kind=role, freshness_version=correlation_id))
            self._record_mutation(actor, action="membership.write", subject=user_id, tenant=tenant,
                                  correlation_id=correlation_id, decision_id=None,
                                  event_type="membership.role_changed", aggregate_id=user_id,
                                  payload={"user_id":user_id, "tenant":tenant, "role":role})

    def require_admin_successor(self, user_id: str, *, tenant: str) -> None:
        """Do not deactivate the last active tenant administrator."""
        members = self.store.memberships_for_tenant(tenant)
        admins = {m.user_id for m in members if m.scope_type == "tenant"
                  and m.scope_id == tenant and m.kind == "tenant-admin"}
        if user_id not in admins:
            return
        account = self.store.tenant_account(tenant, user_id)
        if account is not None and account.status != AccountStatus.ACTIVE:
            return
        for other in admins - {user_id}:
            state = self.store.tenant_account(tenant, other)
            if state is not None and state.status == AccountStatus.ACTIVE:
                return
        raise ConflictError("Assign another active tenant administrator before disabling this account.")

    def set_tenant_account_status(self, actor: Actor, user_id: str, status: AccountStatus, *, tenant: str, correlation_id: str | None = None) -> TenantAccount:
        with self.store.tenant_lifecycle_guard(tenant):
            return self._set_tenant_account_status(actor, user_id, status, tenant=tenant, correlation_id=correlation_id)

    def _set_tenant_account_status(
        self,
        actor: Actor,
        user_id: str,
        status: AccountStatus,
        *,
        tenant: str,
        correlation_id: str | None = None,
    ) -> TenantAccount:
        self._require_user(user_id)
        tenant_context = self.resolve_tenant_context(actor, tenant)
        correlation_id = correlation_id or new_id("corr")
        account = self.store.tenant_account(tenant_context.tenant, user_id)
        if account is None:
            account = TenantAccount(user_id=user_id, tenant=tenant_context.tenant)
        decision = self._authorize(
            actor,
            action="tenant.account.update",
            resource_type="user-engine:tenant-account",
            resource_id=f"{tenant_context.tenant}:{user_id}",
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            target_user_id=user_id,
        )
        updated = replace(account, status=status)
        with self.store.transaction():
            if status in {AccountStatus.SUSPENDED, AccountStatus.DISABLED}:
                self.require_admin_successor(user_id, tenant=tenant_context.tenant)
            self.store.save_tenant_account(updated)
            self._record_mutation(
                actor,
                action="tenant.account.update",
                subject=user_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="tenant_account.status_changed",
                aggregate_id=user_id,
                payload={
                    "user_id": user_id,
                    "tenant": tenant_context.tenant,
                    "status": status,
                },
            )
        return updated

    def add_membership(
        self,
        actor: Actor,
        user_id: str,
        *,
        tenant: str,
        scope_type: str,
        scope_id: str,
        kind: str,
        correlation_id: str | None = None,
    ) -> Membership:
        self._require_user(user_id)
        tenant_context = self.resolve_tenant_context(actor, tenant)
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="membership.write",
            resource_type="user-engine:membership",
            resource_id=f"{tenant_context.tenant}:{user_id}:{scope_type}:{scope_id}",
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            target_user_id=user_id,
            context={"scope_type": scope_type, "scope_id": scope_id, "kind": kind},
        )
        membership = Membership(
            membership_id=new_id("mem"),
            user_id=user_id,
            tenant=tenant_context.tenant,
            scope_type=scope_type,
            scope_id=scope_id,
            kind=kind,
            freshness_version=correlation_id,
        )
        with self.store.transaction():
            self.store.save_membership(membership)
            self._record_mutation(
                actor,
                action="membership.write",
                subject=user_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="membership.added",
                aggregate_id=membership.membership_id,
                payload={
                    "membership_id": membership.membership_id,
                    "user_id": user_id,
                    "tenant": tenant_context.tenant,
                    "scope_type": scope_type,
                    "scope_id": scope_id,
                    "kind": kind,
                },
            )
        return membership

    def link_identity(
        self,
        actor: Actor,
        user_id: str,
        *,
        issuer: str,
        subject: str,
        provider: str | None = None,
        correlation_id: str | None = None,
    ) -> ExternalIdentity:
        self._require_user(user_id)
        existing = self.store.find_identity(issuer, subject)
        if existing is not None and existing.user_id != user_id:
            raise ConflictError("external identity is already linked")
        if existing is not None:
            return existing

        correlation_id = correlation_id or new_id("corr")
        actor_identity = self.store.find_identity(*actor.identity_key)
        decision = self._authorize(
            actor,
            action="identity.link",
            resource_type="user-engine:identity",
            resource_id=f"{issuer}:{subject}",
            tenant=actor.tenant,
            correlation_id=correlation_id,
            target_user_id=user_id,
            context={
                "self": actor_identity is not None
                and actor_identity.user_id == user_id,
            },
        )
        identity = ExternalIdentity(
            identity_id=new_id("idn"),
            user_id=user_id,
            issuer=issuer,
            subject=subject,
            provider=provider,
        )
        with self.store.transaction():
            self.store.save_identity(identity)
            self._record_mutation(
                actor,
                action="identity.link",
                subject=user_id,
                tenant=actor.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="identity.linked",
                aggregate_id=user_id,
                payload={"issuer": issuer, "subject": subject, "user_id": user_id},
            )
        return identity

    def register_application(
        self,
        actor: Actor,
        application: Application,
        *,
        binding: ApplicationBinding | None = None,
        correlation_id: str | None = None,
    ) -> Application:
        if self.store.application(application.application_id) is not None:
            raise ConflictError("application already exists")
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="application.register",
            resource_type="user-engine:application",
            resource_id=application.application_id,
            tenant=actor.tenant,
            correlation_id=correlation_id,
            application_id=application.application_id,
        )
        with self.store.transaction():
            self.store.save_application(application)
            if binding is not None:
                if binding.application_id != application.application_id:
                    raise ValidationError("binding application id must match application")
                self.store.save_binding(binding)
            self._record_mutation(
                actor,
                action="application.register",
                subject=application.application_id,
                tenant=actor.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="application.registered",
                aggregate_id=application.application_id,
                payload={"application_id": application.application_id},
                application_id=application.application_id,
            )
        return application

    def publish_catalog(
        self,
        actor: Actor,
        catalog: Catalog,
        *,
        correlation_id: str | None = None,
    ) -> Catalog:
        self._validate_catalog(catalog)
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="catalog.publish",
            resource_type="user-engine:catalog",
            resource_id=catalog.catalog_id,
            tenant=actor.tenant,
            correlation_id=correlation_id,
            application_id=catalog.owning_application_id,
        )
        with self.store.transaction():
            self.store.save_catalog(catalog)
            self._record_mutation(
                actor,
                action="catalog.publish",
                subject=catalog.catalog_id,
                tenant=actor.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="catalog.published",
                aggregate_id=catalog.catalog_id,
                payload={
                    "catalog_id": catalog.catalog_id,
                    "namespace": catalog.namespace,
                    "version": catalog.version,
                },
                application_id=catalog.owning_application_id,
            )
        return catalog

    def set_profile_value(
        self,
        actor: Actor,
        user_id: str,
        attribute_key: str,
        value: Any,
        *,
        scope: ProfileScope = ProfileScope.GLOBAL,
        scope_id: str | None = None,
        tenant: str | None = None,
        application_id: str | None = None,
        correlation_id: str | None = None,
    ) -> ProfileValue:
        self._require_user(user_id)
        definition = self._require_attribute(attribute_key)
        self._validate_profile_scope(definition, scope, scope_id)
        self._validate_value(definition, value)
        operation_tenant = self._operation_tenant(actor, tenant, scope, scope_id)
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="profile.write",
            resource_type="user-engine:profile",
            resource_id=f"{user_id}:{attribute_key}",
            tenant=operation_tenant,
            correlation_id=correlation_id,
            application_id=application_id,
            target_user_id=user_id,
            context={"scope": scope, "scope_id": scope_id},
        )
        profile_value = ProfileValue(
            user_id=user_id,
            attribute_key=attribute_key,
            value=value,
            scope=scope,
            scope_id=scope_id,
            source=f"{actor.issuer}:{actor.subject}",
        )
        with self.store.transaction():
            self.store.save_profile_value(profile_value)
            self._record_mutation(
                actor,
                action="profile.write",
                subject=user_id,
                tenant=operation_tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="profile.value_set",
                aggregate_id=user_id,
                payload={
                    "user_id": user_id,
                    "attribute_key": attribute_key,
                    "scope": scope,
                    "scope_id": scope_id,
                },
                application_id=application_id,
            )
        return profile_value

    def effective_profile(
        self,
        actor: Actor,
        user_id: str,
        *,
        application_id: str | None = None,
        tenant: str | None = None,
        correlation_id: str | None = None,
    ) -> EffectiveProfile:
        tenant_context = self.resolve_tenant_context(actor, tenant)
        correlation_id = correlation_id or new_id("corr")
        self._authorize(
            actor,
            action="profile.read",
            resource_type="user-engine:profile",
            resource_id=user_id,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            application_id=application_id,
            target_user_id=user_id,
        )
        return self._resolve_effective_profile(user_id, application_id, tenant_context.tenant)

    def projection(
        self,
        actor: Actor,
        user_id: str,
        projection_type: ProjectionType,
        *,
        application_id: str | None = None,
        tenant: str | None = None,
        correlation_id: str | None = None,
    ) -> Projection:
        if _is_application_projection(projection_type) and application_id is None:
            raise ValidationError("application projections require application_id")
        tenant_context = self.resolve_tenant_context(actor, tenant)
        correlation_id = correlation_id or new_id("corr")
        self._authorize(
            actor,
            action="projection.read",
            resource_type="user-engine:projection",
            resource_id=user_id,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            application_id=application_id,
            target_user_id=user_id,
            context={"projection_type": projection_type},
        )
        effective = self._resolve_effective_profile(
            user_id, application_id, tenant_context.tenant
        )
        owning_application_id = (
            application_id if _is_application_projection(projection_type) else None
        )
        values: dict[str, Any] = {}
        redactions: dict[str, str] = {}
        for key, definition in self._active_attribute_definitions(
            owning_application_id=owning_application_id
        ).items():
            if key not in effective.values:
                continue
            if not _visible_in_projection(definition, projection_type):
                continue
            if _must_redact(definition, projection_type):
                values[key] = REDACTED
                redactions[key] = "sensitivity"
            else:
                values[key] = effective.values[key]
        return Projection(
            user_id=user_id,
            projection_type=projection_type,
            application_id=application_id,
            values=values,
            redactions=redactions,
            access_context=self._projection_access_context(
                user_id,
                tenant_context.tenant,
                application_id=application_id,
            ),
        )

    def identity_context(
        self,
        actor: Actor,
        *,
        user_id: str | None = None,
        tenant: str | None = None,
        application_id: str | None = None,
        include_profile: bool = False,
        correlation_id: str | None = None,
    ) -> IdentityContext:
        tenant_context = self.resolve_tenant_context(actor, tenant)
        identity = self.store.find_identity(*actor.identity_key)
        if user_id is None:
            if identity is None:
                raise NotFoundError("identity link not found")
            user_id = identity.user_id
        user = self._require_user(user_id)
        account = self._require_account(user_id)
        correlation_id = correlation_id or new_id("corr")
        self._authorize(
            actor,
            action="identity_context.read",
            resource_type="user-engine:identity-context",
            resource_id=user_id,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            application_id=application_id,
            target_user_id=user_id,
            context={
                "include_profile": include_profile,
                "self": identity is not None and identity.user_id == user_id,
            },
        )
        identities = self.store.identities_for_user(user.user_id)
        memberships = self.store.memberships_for_user(
            user.user_id, tenant=tenant_context.tenant
        )
        active_access_context = self.store.active_access_context(
            user.user_id, tenant_context.tenant
        )
        evidence_refs = self._identity_evidence_refs(
            user,
            account,
            memberships,
            tenant_context.tenant,
            active_access_context=active_access_context,
        )
        access_control_facts = self._access_control_facts(
            tenant_context.tenant,
            user_id=user.user_id,
            memberships=memberships,
            active_context=active_access_context,
        )
        onboarding_journeys = self.store.onboarding_journeys_for_user(
            user.user_id, tenant=tenant_context.tenant
        )
        entity_refs = self._identity_entity_refs(
            actor,
            user,
            account,
            identities,
            memberships,
            tenant_context.tenant,
            application_id,
            active_access_context,
        )
        relationship_refs, grant_like_refs = self._identity_relationship_refs(
            entity_refs,
            memberships,
            tenant_context.tenant,
            evidence_refs[0].identifier if evidence_refs else None,
            active_access_context=active_access_context,
        )
        profile = (
            self._resolve_effective_profile(
                user.user_id, application_id, tenant_context.tenant
            )
            if include_profile
            else None
        )
        gaps = () if evidence_refs else ("evidence:no-audit-records",)
        return IdentityContext(
            actor=actor,
            user=user,
            account=account,
            identities=identities,
            tenant=tenant_context.tenant,
            application_id=application_id,
            memberships=memberships,
            entity_refs=entity_refs,
            relationship_refs=relationship_refs,
            grant_like_refs=grant_like_refs,
            evidence_refs=evidence_refs,
            profile=profile,
            gaps=gaps,
            active_access_context=active_access_context,
            access_control_facts=access_control_facts,
            onboarding_journeys=onboarding_journeys,
        )

    def onboard_family_dataspace(
        self,
        actor: Actor,
        request: FamilyDataspaceRequest,
        *,
        correlation_id: str | None = None,
    ) -> FamilyDataspaceOnboarding:
        tenant_context = self.resolve_tenant_context(actor, request.tenant)
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="family_dataspace.onboard",
            resource_type="user-engine:family-dataspace",
            resource_id=request.family_scope_id,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            application_id=request.application_id,
            context={"family_scope_id": request.family_scope_id},
        )
        with self.store.transaction():
            owner_session = self._ensure_actor_session(actor, correlation_id)
            application, binding = self._ensure_family_dataspace_application(
                actor, request, correlation_id
            )
            catalog = self._ensure_family_dataspace_catalog(
                actor, request, correlation_id
            )
            owner_membership = self._ensure_membership(
                actor,
                owner_session.user.user_id,
                tenant=tenant_context.tenant,
                scope_type="family",
                scope_id=request.family_scope_id,
                kind=FamilyRole.OWNER.value,
                correlation_id=correlation_id,
            )
            owner_defaults = dict(request.owner_profile_defaults)
            owner_defaults.setdefault(
                "member_display_name",
                owner_session.user.display_name
                or actor.preferred_username
                or owner_session.user.primary_email
                or actor.subject,
            )
            self._apply_family_profile_defaults(
                actor,
                owner_session.user.user_id,
                tenant=tenant_context.tenant,
                application_id=request.application_id,
                catalog_namespace=request.catalog_namespace,
                values=owner_defaults,
                correlation_id=correlation_id,
            )
            invitations = tuple(
                self.invite_family_member(
                    actor,
                    tenant=tenant_context.tenant,
                    family_scope_id=request.family_scope_id,
                    application_id=request.application_id,
                    catalog_namespace=request.catalog_namespace,
                    member=member,
                    correlation_id=correlation_id,
                )
                for member in request.member_specs
            )
            self._record_mutation(
                actor,
                action="family_dataspace.onboard",
                subject=request.family_scope_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="family_dataspace.onboarded",
                aggregate_id=request.family_scope_id,
                payload={
                    "tenant": tenant_context.tenant,
                    "family_scope_id": request.family_scope_id,
                    "application_id": request.application_id,
                    "member_invitation_count": len(invitations),
                },
                application_id=request.application_id,
            )
        identity_context = self.identity_context(
            actor,
            user_id=owner_session.user.user_id,
            tenant=tenant_context.tenant,
            application_id=request.application_id,
            include_profile=True,
            correlation_id=correlation_id,
        )
        claims_projection = self.projection(
            actor,
            owner_session.user.user_id,
            ProjectionType.CLAIMS_ENRICHMENT,
            tenant=tenant_context.tenant,
            application_id=request.application_id,
            correlation_id=correlation_id,
        )
        return FamilyDataspaceOnboarding(
            tenant=tenant_context.tenant,
            family_scope_id=request.family_scope_id,
            family_display_name=request.family_display_name,
            owner_session=owner_session,
            owner_membership=owner_membership,
            application=application,
            binding=binding,
            catalog=catalog,
            invitations=invitations,
            identity_context=identity_context,
            claims_projection=claims_projection,
        )

    def invite_family_member(
        self,
        actor: Actor,
        *,
        tenant: str,
        family_scope_id: str,
        application_id: str,
        member: FamilyMemberSpec,
        catalog_namespace: str = "dataspace",
        correlation_id: str | None = None,
        expires_in: timedelta = timedelta(days=7),
    ) -> FamilyMemberInvitation:
        tenant_context = self.resolve_tenant_context(actor, tenant)
        correlation_id = correlation_id or new_id("corr")
        role = _family_role_value(member.role)
        if not member.primary_email:
            raise ValidationError("family member primary_email is required")
        if any(
            item.primary_email.casefold() == member.primary_email.casefold()
            and item.status == InvitationStatus.PENDING
            for item in self.store.family_invitations_for_tenant(tenant_context.tenant)
        ):
            raise ConflictError("a pending invitation already exists for this address")
        if member.issuer and member.subject:
            existing = self.store.find_identity(member.issuer, member.subject)
            if existing is not None:
                raise ConflictError("external identity is already linked")
        decision = self._authorize(
            actor,
            action="family_member.invite",
            resource_type="user-engine:family-invitation",
            resource_id=member.primary_email,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            application_id=application_id,
            context={
                "family_scope_id": family_scope_id,
                "role": role,
            },
        )
        with self.store.transaction():
            user = self.create_user(
                actor,
                display_name=member.display_name,
                primary_email=member.primary_email,
                correlation_id=correlation_id,
            )
            tenant_account = self.set_tenant_account_status(
                actor,
                user.user_id,
                AccountStatus.INVITED,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
            )
            membership = self._ensure_membership(
                actor,
                user.user_id,
                tenant=tenant_context.tenant,
                scope_type="family",
                scope_id=family_scope_id,
                kind=role,
                correlation_id=correlation_id,
            )
            profile_defaults = dict(member.profile_defaults)
            if member.display_name and role not in {"user", TENANT_ADMIN_ROLE}:
                profile_defaults.setdefault("member_display_name", member.display_name)
            self._apply_family_profile_defaults(
                actor,
                user.user_id,
                tenant=tenant_context.tenant,
                application_id=application_id,
                catalog_namespace=catalog_namespace,
                values=profile_defaults,
                correlation_id=correlation_id,
            )
            if member.issuer and member.subject:
                self.link_identity(
                    actor,
                    user.user_id,
                    issuer=member.issuer,
                    subject=member.subject,
                    provider=member.provider,
                    correlation_id=correlation_id,
                )
            invitation = FamilyInvitation(
                invitation_id=new_id("inv"),
                tenant=tenant_context.tenant,
                family_scope_id=family_scope_id,
                application_id=application_id,
                user_id=user.user_id,
                primary_email=member.primary_email,
                role=role,
                invited_by=actor.subject,
                correlation_id=correlation_id,
                last_sent_correlation_id=correlation_id,
                expires_at=utc_now() + expires_in,
            )
            self.store.save_family_invitation(invitation)
            self._record_mutation(
                actor,
                action="family_member.invite",
                subject=invitation.invitation_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="family_member.invited",
                aggregate_id=invitation.invitation_id,
                payload={
                    "invitation_id": invitation.invitation_id,
                    "user_id": user.user_id,
                    "tenant": tenant_context.tenant,
                    "family_scope_id": family_scope_id,
                    "application_id": application_id,
                    "role": role,
                    "primary_email": invitation.primary_email,
                },
                application_id=application_id,
            )
        return FamilyMemberInvitation(
            user=user,
            tenant_account=tenant_account,
            membership=membership,
            invitation=invitation,
        )

    def resend_family_invitation(
        self,
        actor: Actor,
        invitation_id: str,
        *,
        correlation_id: str | None = None,
        expected_version: int | None = None,
    ) -> FamilyInvitation:
        invitation = self._require_family_invitation(invitation_id)
        if invitation.status != InvitationStatus.PENDING:
            raise ValidationError("only pending invitations can be resent")
        if expected_version is not None and invitation.version != expected_version:
            raise ConflictError("invitation version does not match")
        tenant_context = self.resolve_tenant_context(actor, invitation.tenant)
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="family_invitation.resend",
            resource_type="user-engine:family-invitation",
            resource_id=invitation.invitation_id,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            application_id=invitation.application_id,
            target_user_id=invitation.user_id,
        )
        updated = replace(
            invitation,
            resend_count=invitation.resend_count + 1,
            last_sent_correlation_id=correlation_id,
            updated_at=utc_now(),
            expires_at=utc_now() + timedelta(days=7),
            version=invitation.version + 1,
        )
        with self.store.transaction():
            self.store.save_family_invitation(updated)
            self._record_mutation(
                actor,
                action="family_invitation.resend",
                subject=updated.invitation_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="family_invitation.resent",
                aggregate_id=updated.invitation_id,
                payload={
                    "invitation_id": updated.invitation_id,
                    "user_id": updated.user_id,
                    "resend_count": updated.resend_count,
                    "primary_email": updated.primary_email,
                },
                application_id=updated.application_id,
            )
        return updated

    def revoke_family_invitation(
        self,
        actor: Actor,
        invitation_id: str,
        *,
        correlation_id: str | None = None,
        expected_version: int | None = None,
    ) -> FamilyInvitation:
        invitation = self._require_family_invitation(invitation_id)
        if invitation.status != InvitationStatus.PENDING:
            raise ValidationError("only pending invitations can be revoked")
        if expected_version is not None and invitation.version != expected_version:
            raise ConflictError("invitation version does not match")
        tenant_context = self.resolve_tenant_context(actor, invitation.tenant)
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="family_invitation.revoke",
            resource_type="user-engine:family-invitation",
            resource_id=invitation.invitation_id,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            application_id=invitation.application_id,
            target_user_id=invitation.user_id,
        )
        with self.store.transaction():
            self.set_tenant_account_status(
                actor,
                invitation.user_id,
                AccountStatus.DISABLED,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
            )
            updated = replace(
                invitation,
                status=InvitationStatus.REVOKED,
                updated_at=utc_now(),
                revoked_at=utc_now(),
                version=invitation.version + 1,
            )
            self.store.save_family_invitation(updated)
            self._record_mutation(
                actor,
                action="family_invitation.revoke",
                subject=updated.invitation_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="family_invitation.revoked",
                aggregate_id=updated.invitation_id,
                payload={
                    "invitation_id": updated.invitation_id,
                    "user_id": updated.user_id,
                    "status": updated.status,
                },
                application_id=updated.application_id,
            )
        return updated

    def accept_family_invitation(
        self,
        claims: Mapping[str, Any],
        invitation_id: str,
        *,
        correlation_id: str | None = None,
    ) -> FamilyInvitationAcceptance:
        invitation = self._require_family_invitation(invitation_id)
        if invitation.status == InvitationStatus.REVOKED:
            raise ValidationError("revoked invitations cannot be accepted")
        if invitation.status == InvitationStatus.ACCEPTED:
            raise ValidationError("invitation is already accepted")
        if invitation.expires_at is not None and invitation.expires_at <= utc_now():
            raise ValidationError("invitation has expired")
        actor = self.identity_adapter.normalize(claims)
        tenant_context = self.resolve_tenant_context(actor, invitation.tenant)
        claimed_email = str(claims.get("email") or "").strip().casefold()
        if not claimed_email or claimed_email != invitation.primary_email.casefold():
            raise AuthorizationDenied("invitation identity does not match")
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action="family_invitation.accept",
            resource_type="user-engine:family-invitation",
            resource_id=invitation.invitation_id,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            application_id=invitation.application_id,
            target_user_id=invitation.user_id,
            context={"family_scope_id": invitation.family_scope_id},
        )
        with self.store.transaction():
            self.link_identity(
                actor,
                invitation.user_id,
                issuer=actor.issuer,
                subject=actor.subject,
                provider=actor.authorized_party,
                correlation_id=correlation_id,
            )
            account = self.set_account_status(
                actor,
                invitation.user_id,
                AccountStatus.ACTIVE,
                correlation_id=correlation_id,
            )
            self.set_tenant_account_status(
                actor,
                invitation.user_id,
                AccountStatus.ACTIVE,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
            )
            accepted_at = utc_now()
            accepted = replace(
                invitation,
                status=InvitationStatus.ACCEPTED,
                updated_at=accepted_at,
                accepted_at=accepted_at,
                version=invitation.version + 1,
            )
            self.store.save_family_invitation(accepted)
            self._record_mutation(
                actor,
                action="family_invitation.accept",
                subject=accepted.invitation_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="family_invitation.accepted",
                aggregate_id=accepted.invitation_id,
                payload={
                    "invitation_id": accepted.invitation_id,
                    "user_id": accepted.user_id,
                    "tenant": accepted.tenant,
                    "family_scope_id": accepted.family_scope_id,
                    "application_id": accepted.application_id,
                    "status": accepted.status,
                },
                application_id=accepted.application_id,
            )
        session = self._session(actor, self._require_user(accepted.user_id), account)
        identity_context = self.identity_context(
            actor,
            user_id=accepted.user_id,
            tenant=tenant_context.tenant,
            application_id=accepted.application_id,
            include_profile=True,
            correlation_id=correlation_id,
        )
        claims_projection = self.projection(
            actor,
            accepted.user_id,
            ProjectionType.CLAIMS_ENRICHMENT,
            tenant=tenant_context.tenant,
            application_id=accepted.application_id,
            correlation_id=correlation_id,
        )
        return FamilyInvitationAcceptance(
            session=session,
            invitation=accepted,
            identity_context=identity_context,
            claims_projection=claims_projection,
        )

    def tenant_diagnostics(
        self,
        actor: Actor,
        *,
        tenant: str,
        correlation_id: str | None = None,
    ) -> TenantDiagnostics:
        tenant_context = self.resolve_tenant_context(actor, tenant)
        correlation_id = correlation_id or new_id("corr")
        self._authorize(
            actor,
            action="tenant.diagnostics.read",
            resource_type="user-engine:tenant",
            resource_id=tenant_context.tenant,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
        )
        memberships = self.store.memberships_for_tenant(tenant_context.tenant)
        checks = {
            "store_schema": self.store.ready,
            "authorization_port": self.authorization is not None,
            "membership_facts": bool(memberships),
            "catalog_scopes": bool(self._active_attribute_definitions()),
            "audit_ready": self.store.ready,
        }
        issues = tuple(key for key, passed in checks.items() if not passed)
        return TenantDiagnostics(
            tenant=tenant_context.tenant,
            checks=checks,
            issues=issues,
            memberships=memberships,
        )

    def audit_records(self) -> tuple[AuditRecord, ...]:
        return self.store.audit_log()

    def outbox_events(self) -> tuple[OutboxEvent, ...]:
        return self.store.pending_outbox()

    def deliver_outbox(
        self, actor: Actor, deliver: Callable[[OutboxEvent], None], *,
        worker_id: str, max_attempts: int = 3, correlation_id: str | None = None,
        event_id: str | None = None, expected_attempts: int | None = None,
    ) -> tuple[OutboxEvent, ...]:
        """Serialize each external attempt and retain only classified failures."""
        if not worker_id.strip() or not 1 <= max_attempts <= 20:
            raise ValidationError("invalid outbox worker or attempt limit")
        correlation_id = correlation_id or new_id("corr")
        self._authorize(actor, action="outbox.deliver", resource_type="user-engine:outbox",
            resource_id="pending", tenant=actor.tenant, correlation_id=correlation_id)
        selected = self.store.outbox_event(event_id) if event_id else None
        if event_id and selected is None:
            raise NotFoundError("delivery record not found")
        events = (selected,) if selected else self.store.pending_outbox()
        results = []
        for candidate in events:
            with self.store.outbox_delivery_guard(candidate.event_id):
                event = self.store.outbox_event(candidate.event_id)
                if event is None or event.delivered_at is not None:
                    continue
                if expected_attempts is not None and event.delivery_attempts != expected_attempts:
                    raise ConflictError("Delivery changed. Refresh and review the attempt again.")
                # Owning this cross-process guard proves any previous claim is orphaned.
                # Receiver idempotency reconciles a lost receipt without blind resending.
                if event.dead_lettered_at and not event_id:
                    continue
                claimed = replace(event, claimed_by=worker_id, claimed_at=utc_now(),
                    delivery_attempts=event.delivery_attempts+1, failed_at=None, failure_reason=None)
                with self.store.transaction(): self.store.save_outbox(claimed)
                try:
                    deliver(claimed)
                except Exception as exc:
                    code = getattr(exc, "code", "delivery_unavailable")
                    if code not in {"mail_unconfigured", "delivery_outcome_unknown", "temporary_deferral", "permanent_rejection", "recipient_suppressed", "provider_timeout", "provider_unavailable"}:
                        code = "delivery_unavailable"
                    outcome = replace(claimed, claimed_by=None, claimed_at=None,
                        failed_at=utc_now(), failure_reason=code,
                        dead_lettered_at=utc_now() if claimed.delivery_attempts >= max_attempts else None)
                else:
                    outcome = replace(claimed, claimed_by=None, claimed_at=None,
                        delivered_at=utc_now(), dead_lettered_at=None)
                with self.store.transaction(): self.store.save_outbox(outcome)
                results.append(outcome)
        return tuple(results)

    def replay_outbox(self, actor: Actor, event_id: str, *, correlation_id: str | None = None) -> OutboxEvent:
        correlation_id = correlation_id or new_id("corr")
        with self.store.outbox_delivery_guard(event_id):
            event = self.store.outbox_event(event_id)
            if event is None: raise NotFoundError("outbox event not found")
            self.resolve_tenant_context(actor, event.tenant)
            self._authorize(actor, action="outbox.replay", resource_type="user-engine:outbox-event",
                resource_id=event_id, tenant=event.tenant, correlation_id=correlation_id)
            if event.delivered_at or (event.claimed_by and not event.failed_at):
                raise ConflictError("Delivery is completed or its previous attempt is unresolved.")
            replayed = replace(event, claimed_by=None, claimed_at=None, failed_at=None,
                failure_reason=None, dead_lettered_at=None)
            with self.store.transaction(): self.store.save_outbox(replayed)
            return replayed

    def outbox_diagnostics(self) -> OutboxDiagnostics:
        event_types: dict[str, int] = {}
        pending = self.store.pending_outbox()
        for event in pending:
            event_types[event.event_type] = event_types.get(event.event_type, 0) + 1
        oldest = pending[0].correlation_id if pending else None
        return OutboxDiagnostics(
            pending_count=len(pending),
            event_types=event_types,
            oldest_correlation_id=oldest,
        )

    def operability_snapshot(self) -> OperabilitySnapshot:
        audit_correlation_ok = all(
            bool(record.correlation_id) for record in self.store.audit_log()
        )
        checks = {
            "ready": self.readiness().ready,
            "audit_correlation": audit_correlation_ok,
            "outbox_diagnostics": self.outbox_diagnostics().pending_count >= 0,
        }
        metrics = self.store.record_counts()
        issues = tuple(key for key, passed in checks.items() if not passed)
        return OperabilitySnapshot(
            ready=all(checks.values()),
            checks=checks,
            metrics=metrics,
            issues=issues,
        )

    def record_evidence_heartbeat(
        self, *, correlation_id: str | None = None
    ) -> OutboxEvent:
        """Emit a positive evidence claim that can itself go missing.

        Counts load-bearing denials and revocations already on the local
        trail. Does not claim the trail is complete.
        """
        correlation_id = correlation_id or new_id("corr")
        counts = load_bearing_counts(
            self.store.audit_log(), self.store.pending_outbox()
        )
        event = OutboxEvent(
            event_id=new_id("evt"),
            event_type=HEARTBEAT_EVENT_TYPE,
            aggregate_id="user-engine:evidence",
            payload={
                "completeness_claimed": False,
                "form": "heartbeat",
                "bound": (
                    "the archive proves the records it holds were not "
                    "altered or truncated after arrival"
                ),
                "load_bearing": counts,
            },
            tenant=PLATFORM_TENANT,
            correlation_id=correlation_id,
        )
        self.store.append_outbox(event)
        return event

    def structured_log_context(
        self,
        *,
        correlation_id: str,
        tenant: str,
        actor: Actor | None = None,
    ) -> Mapping[str, str]:
        context = {
            "correlation_id": correlation_id,
            "tenant": tenant,
        }
        if actor is not None:
            context["actor_issuer"] = actor.issuer
            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 _close_prepared_account(
        self,
        actor: Actor,
        prepared_account_id: str,
        *,
        status: PreparedAccountStatus,
        action: str,
        event_type: str,
        correlation_id: str | None,
    ) -> PreparedAccount:
        prepared = self._require_prepared_account(prepared_account_id)
        self._ensure_prepared_account_pending(prepared)
        tenant_context = self.resolve_tenant_context(actor, prepared.tenant)
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action=action,
            resource_type="user-engine:prepared-account",
            resource_id=prepared.prepared_account_id,
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
        )
        now = utc_now()
        updated = replace(
            prepared,
            status=status,
            updated_at=now,
            revoked_at=now if status == PreparedAccountStatus.REVOKED else None,
            expired_at=now if status == PreparedAccountStatus.EXPIRED else None,
        )
        with self.store.transaction():
            self.store.save_prepared_account(updated)
            self._record_mutation(
                actor,
                action=action,
                subject=prepared.prepared_account_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type=event_type,
                aggregate_id=prepared.prepared_account_id,
                payload={
                    "prepared_account_id": prepared.prepared_account_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 _require_prepared_account(self, prepared_account_id: str) -> PreparedAccount:
        prepared = self.store.prepared_account(prepared_account_id)
        if prepared is None:
            raise NotFoundError("prepared account not found")
        return prepared

    def _ensure_prepared_account_pending(self, prepared: PreparedAccount) -> None:
        if prepared.status != PreparedAccountStatus.PENDING:
            raise ValidationError("prepared account is not pending")

    def _ensure_no_duplicate_prepared_account(
        self,
        tenant: str,
        requirements: tuple[PreparedFactorRequirement, ...],
        *,
        excluding_prepared_account_id: str | None,
    ) -> None:
        signature = _prepared_factor_signature(requirements)
        for prepared in self.store.prepared_accounts_for_tenant(tenant):
            if prepared.prepared_account_id == excluding_prepared_account_id:
                continue
            if prepared.status != PreparedAccountStatus.PENDING:
                continue
            if _prepared_account_expired(prepared):
                continue
            if _prepared_factor_signature(prepared.required_factor_matches) == signature:
                raise ConflictError("matching prepared account already exists")

    def _validate_prepared_factor_requirements(
        self, requirements: tuple[PreparedFactorRequirement, ...]
    ) -> None:
        if not requirements:
            raise ValidationError("prepared account requires at least one factor match")
        for requirement in requirements:
            if not requirement.normalized_value.strip():
                raise ValidationError(
                    "prepared account factor match requires normalized value"
                )

    def _validate_prepared_entitlements(
        self, entitlements: tuple[PreparedEntitlement, ...], tenant: str
    ) -> None:
        for entitlement in entitlements:
            if entitlement.tenant != tenant:
                raise ValidationError("prepared entitlement tenant must match account")
            if entitlement.kind == PreparedEntitlementKind.TENANT_ACCOUNT:
                continue
            if entitlement.kind == PreparedEntitlementKind.MEMBERSHIP:
                if not entitlement.scope_type or not entitlement.scope_id:
                    raise ValidationError("membership entitlement requires scope")
                if not entitlement.role:
                    raise ValidationError("membership entitlement requires role")
            elif entitlement.kind == PreparedEntitlementKind.PROFILE_VALUE:
                if not entitlement.attribute_key:
                    raise ValidationError("profile entitlement requires attribute_key")
            elif entitlement.kind == PreparedEntitlementKind.APPLICATION_BINDING:
                if entitlement.application_binding is None:
                    raise ValidationError(
                        "application binding entitlement requires binding"
                    )
            elif entitlement.kind == PreparedEntitlementKind.ONBOARDING_JOURNEY:
                if not entitlement.onboarding_journey:
                    raise ValidationError(
                        "onboarding journey entitlement requires journey"
                    )
            else:
                raise ValidationError(f"unsupported entitlement kind: {entitlement.kind}")

    def _resolve_prepared_account_claim(
        self,
        actor: Actor,
        session: RegistrationSession,
        factors: tuple[IdentityFactor, ...],
        *,
        prepared_account_id: str | None,
        tenant: str,
        correlation_id: str,
    ) -> PreparedAccount:
        candidates = (
            (self._require_prepared_account(prepared_account_id),)
            if prepared_account_id is not None
            else self.store.prepared_accounts_for_tenant(tenant)
        )
        matches = tuple(
            prepared
            for prepared in candidates
            if prepared.tenant == tenant
            and prepared.status == PreparedAccountStatus.PENDING
            and not _prepared_account_expired(prepared)
            and _prepared_account_matches_factors(prepared, factors)
        )
        if prepared_account_id is not None and not matches:
            self._record_audit_only(
                actor,
                action="prepared_account.claim",
                subject=prepared_account_id,
                tenant=tenant,
                correlation_id=correlation_id,
                summary="prepared account claim denied: factor mismatch or closed",
            )
            raise ValidationError("prepared account claim requirements are not met")
        if not matches:
            self._record_audit_only(
                actor,
                action="prepared_account.claim",
                subject=session.registration_id,
                tenant=tenant,
                correlation_id=correlation_id,
                summary="prepared account claim denied: no match",
            )
            raise NotFoundError("no matching prepared account found")
        if len(matches) > 1:
            self._record_audit_only(
                actor,
                action="prepared_account.claim",
                subject=session.registration_id,
                tenant=tenant,
                correlation_id=correlation_id,
                summary="prepared account claim denied: ambiguous match",
            )
            raise ConflictError("multiple prepared accounts match registration")
        return matches[0]

    def _prepared_entitlement_records(
        self,
        prepared: PreparedAccount,
        user_id: str,
        *,
        correlation_id: str,
    ) -> tuple[
        tuple[TenantAccount, ...],
        tuple[Membership, ...],
        tuple[ProfileValue, ...],
        tuple[ApplicationBinding, ...],
        tuple[str, ...],
    ]:
        tenant_accounts: list[TenantAccount] = []
        memberships: list[Membership] = []
        profile_values: list[ProfileValue] = []
        application_bindings: list[ApplicationBinding] = []
        onboarding_journeys: list[str] = []
        for entitlement in prepared.entitlements:
            if entitlement.kind == PreparedEntitlementKind.TENANT_ACCOUNT:
                tenant_accounts.append(
                    TenantAccount(
                        user_id=user_id,
                        tenant=entitlement.tenant,
                        status=entitlement.tenant_account_status,
                    )
                )
            elif entitlement.kind == PreparedEntitlementKind.MEMBERSHIP:
                memberships.append(
                    Membership(
                        membership_id=new_id("mem"),
                        user_id=user_id,
                        tenant=entitlement.tenant,
                        scope_type=entitlement.scope_type or "tenant",
                        scope_id=entitlement.scope_id or entitlement.tenant,
                        kind=entitlement.role or "member",
                        source_system="prepared-account",
                        freshness_version=correlation_id,
                    )
                )
            elif entitlement.kind == PreparedEntitlementKind.PROFILE_VALUE:
                definition = self._require_attribute(entitlement.attribute_key or "")
                self._validate_profile_scope(
                    definition,
                    entitlement.profile_scope,
                    entitlement.profile_scope_id,
                )
                self._validate_value(definition, entitlement.value)
                profile_values.append(
                    ProfileValue(
                        user_id=user_id,
                        attribute_key=entitlement.attribute_key or "",
                        value=entitlement.value,
                        scope=entitlement.profile_scope,
                        scope_id=entitlement.profile_scope_id,
                        source=f"prepared-account:{prepared.prepared_account_id}",
                    )
                )
            elif entitlement.kind == PreparedEntitlementKind.APPLICATION_BINDING:
                if entitlement.application_binding is not None:
                    if self.store.application(
                        entitlement.application_binding.application_id
                    ) is None:
                        raise ValidationError(
                            "application binding entitlement requires registered application"
                        )
                    application_bindings.append(entitlement.application_binding)
            elif entitlement.kind == PreparedEntitlementKind.ONBOARDING_JOURNEY:
                if entitlement.onboarding_journey:
                    onboarding_journeys.append(entitlement.onboarding_journey)
        return (
            tuple(tenant_accounts),
            tuple(memberships),
            tuple(profile_values),
            tuple(application_bindings),
            tuple(onboarding_journeys),
        )

    def _record_audit_only(
        self,
        actor: Actor,
        *,
        action: str,
        subject: str,
        tenant: str,
        correlation_id: str,
        summary: str,
    ) -> None:
        self.store.append_audit(
            AuditRecord(
                audit_id=new_id("aud"),
                actor=actor,
                action=action,
                subject=subject,
                tenant=tenant,
                correlation_id=correlation_id,
                summary=summary,
            )
        )

    def _require_welcome_protocol(self, protocol_id: str) -> WelcomeProtocol:
        protocol = self.store.welcome_protocol(protocol_id)
        if protocol is None:
            raise NotFoundError("welcome protocol not found")
        return protocol

    def _require_onboarding_journey(self, journey_id: str) -> OnboardingJourney:
        journey = self.store.onboarding_journey(journey_id)
        if journey is None:
            raise NotFoundError("onboarding journey not found")
        return journey

    def _validate_welcome_protocol(
        self, protocol: WelcomeProtocol, tenant: str
    ) -> None:
        if protocol.tenant != tenant:
            raise ValidationError("welcome protocol tenant must match context")
        if not protocol.name.strip():
            raise ValidationError("welcome protocol name is required")
        if not protocol.steps:
            raise ValidationError("welcome protocol requires at least one step")
        step_keys = [step.step_key for step in protocol.steps]
        if len(step_keys) != len(set(step_keys)):
            raise ValidationError("welcome protocol step keys must be unique")
        for step in protocol.steps:
            if not step.step_key.strip() or not step.title.strip():
                raise ValidationError("welcome protocol steps require key and title")
            if not step.subsystem.strip():
                raise ValidationError("welcome protocol steps require subsystem")

    def _build_onboarding_journey(
        self,
        protocol: WelcomeProtocol,
        user_id: str,
        *,
        trigger_type: OnboardingTriggerType,
        source_id: str | None,
        source_event_type: str | None,
        correlation_id: str,
    ) -> OnboardingJourney:
        if trigger_type != OnboardingTriggerType.MANUAL and (
            protocol.trigger_type != trigger_type
        ):
            raise ValidationError("welcome protocol trigger does not match journey")
        steps = tuple(_onboarding_step_from_template(step) for step in protocol.steps)
        steps = _activate_initial_onboarding_step(steps)
        status = _journey_status_from_steps(steps)
        return OnboardingJourney(
            tenant=protocol.tenant,
            user_id=user_id,
            protocol_id=protocol.protocol_id,
            trigger_type=trigger_type,
            steps=steps,
            status=status,
            source_id=source_id,
            source_event_type=source_event_type,
            journey_key=protocol.journey_key or protocol.prepared_journey,
            active_step_key=_active_onboarding_step_key(steps),
            correlation_id=correlation_id,
        )

    def _auto_start_onboarding_journeys(
        self,
        actor: Actor,
        user_id: str,
        protocols: tuple[WelcomeProtocol, ...],
        *,
        trigger_type: OnboardingTriggerType,
        source_id: str,
        source_event_type: str,
        correlation_id: str,
        decision_id: str | None,
    ) -> tuple[OnboardingJourney, ...]:
        journeys = []
        for protocol in protocols:
            if self._onboarding_journey_exists(
                user_id,
                protocol.protocol_id,
                source_id,
            ):
                continue
            journey = self._build_onboarding_journey(
                protocol,
                user_id,
                trigger_type=trigger_type,
                source_id=source_id,
                source_event_type=source_event_type,
                correlation_id=correlation_id,
            )
            self.store.save_onboarding_journey(journey)
            self._record_onboarding_journey_started(
                actor,
                journey,
                decision_id=decision_id,
                correlation_id=correlation_id,
            )
            journeys.append(journey)
        return tuple(journeys)

    def _record_onboarding_journey_started(
        self,
        actor: Actor,
        journey: OnboardingJourney,
        *,
        decision_id: str | None,
        correlation_id: str,
    ) -> None:
        self._record_mutation(
            actor,
            action="onboarding_journey.start",
            subject=journey.journey_id,
            tenant=journey.tenant,
            correlation_id=correlation_id,
            decision_id=decision_id,
            event_type="onboarding_journey.started",
            aggregate_id=journey.user_id,
            payload={
                "journey_id": journey.journey_id,
                "user_id": journey.user_id,
                "protocol_id": journey.protocol_id,
                "trigger_type": journey.trigger_type,
                "status": journey.status,
                "source_id": journey.source_id,
                "source_event_type": journey.source_event_type,
                "active_step_key": journey.active_step_key,
                "lifecycle_gaps": tuple(
                    step.lifecycle_gap for step in journey.steps if step.lifecycle_gap
                ),
            },
        )

    def _matching_welcome_protocols_for_registration(
        self, session: RegistrationSession
    ) -> tuple[WelcomeProtocol, ...]:
        factors = self.store.factors_for_registration(session.registration_id)
        factor_types = {factor.factor_type for factor in factors}
        return tuple(
            protocol
            for protocol in self.store.welcome_protocols_for_tenant(session.tenant)
            if protocol.trigger_type == OnboardingTriggerType.REGISTRATION
            and (
                protocol.registration_status is None
                or protocol.registration_status == session.status
            )
            and set(protocol.required_factor_types).issubset(factor_types)
        )

    def _matching_welcome_protocols_for_prepared_account(
        self, prepared: PreparedAccount
    ) -> tuple[WelcomeProtocol, ...]:
        prepared_journeys = {
            entitlement.onboarding_journey
            for entitlement in prepared.entitlements
            if entitlement.onboarding_journey
        }
        return tuple(
            protocol
            for protocol in self.store.welcome_protocols_for_tenant(prepared.tenant)
            if protocol.trigger_type == OnboardingTriggerType.PREPARED_ACCOUNT
            and (
                not protocol.prepared_journey
                or protocol.prepared_journey in prepared_journeys
            )
        )

    def _onboarding_journey_exists(
        self,
        user_id: str,
        protocol_id: str,
        source_id: str | None,
    ) -> bool:
        return any(
            journey.protocol_id == protocol_id and journey.source_id == source_id
            for journey in self.store.onboarding_journeys_for_user(user_id)
        )

    def _transition_onboarding_step(
        self,
        actor: Actor,
        journey_id: str,
        step_key: str,
        *,
        status: OnboardingStepStatus,
        action: str,
        event_type: str,
        lifecycle_gap: str | None = None,
        correlation_id: str | None,
    ) -> OnboardingJourney:
        journey = self._require_onboarding_journey(journey_id)
        tenant_context = self.resolve_tenant_context(actor, journey.tenant)
        if journey.status in (
            OnboardingJourneyStatus.COMPLETED,
            OnboardingJourneyStatus.SKIPPED,
        ):
            raise ValidationError("completed onboarding journeys cannot change")
        correlation_id = correlation_id or new_id("corr")
        decision = self._authorize(
            actor,
            action=action,
            resource_type="user-engine:onboarding-step",
            resource_id=f"{journey.journey_id}:{step_key}",
            tenant=tenant_context.tenant,
            correlation_id=correlation_id,
            target_user_id=journey.user_id,
        )
        updated_steps = _replace_onboarding_step_status(
            journey.steps,
            step_key,
            status=status,
            lifecycle_gap=lifecycle_gap,
        )
        updated = _replace_onboarding_journey_steps(journey, updated_steps)
        with self.store.transaction():
            self.store.save_onboarding_journey(updated)
            self._record_mutation(
                actor,
                action=action,
                subject=updated.journey_id,
                tenant=tenant_context.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type=event_type,
                aggregate_id=updated.user_id,
                payload={
                    "journey_id": updated.journey_id,
                    "user_id": updated.user_id,
                    "step_key": step_key,
                    "step_status": status,
                    "journey_status": updated.status,
                    "active_step_key": updated.active_step_key,
                    "lifecycle_gap": lifecycle_gap,
                },
            )
        return updated

    def _require_access_profile(self, access_profile_id: str) -> AccessProfile:
        profile = self.store.access_profile(access_profile_id)
        if profile is None:
            raise NotFoundError("access profile not found")
        return profile

    def _validate_access_profile(self, profile: AccessProfile, tenant: str) -> None:
        if profile.tenant != tenant:
            raise ValidationError("access profile tenant must match context")
        if not profile.display_name.strip():
            raise ValidationError("access profile display_name is required")
        if not profile.hat.strip():
            raise ValidationError("access profile hat is required")
        if profile.scope_type != AccessScopeType.TENANT and not profile.scope_id:
            raise ValidationError("access profile scope_id is required")
        if not profile.membership_requirements:
            raise ValidationError(
                "access profile requires at least one membership requirement"
            )
        for requirement in profile.membership_requirements:
            if (
                not requirement.scope_type.strip()
                or not requirement.scope_id.strip()
                or not requirement.kind.strip()
            ):
                raise ValidationError("membership requirements must be complete")

    def _satisfying_access_memberships(
        self,
        profile: AccessProfile,
        memberships: tuple[Membership, ...],
    ) -> tuple[Membership, ...]:
        matched = []
        for requirement in profile.membership_requirements:
            match = next(
                (
                    membership
                    for membership in memberships
                    if membership.scope_type == requirement.scope_type
                    and membership.scope_id == requirement.scope_id
                    and membership.kind == requirement.kind
                ),
                None,
            )
            if match is None:
                raise ValidationError(
                    "access profile membership requirements are not met"
                )
            matched.append(match)
        return tuple(matched)

    def _verified_factor_ids_for_user(
        self,
        user_id: str,
        required_factor_types: tuple[IdentityFactorType, ...],
    ) -> tuple[str, ...]:
        if not required_factor_types:
            return ()
        now = utc_now()
        factors = self.store.factors_for_user(user_id)
        factor_ids = []
        for required_type in required_factor_types:
            match = next(
                (
                    factor
                    for factor in factors
                    if factor.factor_type == required_type
                    and (factor.expires_at is None or factor.expires_at > now)
                ),
                None,
            )
            if match is None:
                raise ValidationError("access profile factor requirements are not met")
            factor_ids.append(match.factor_id)
        return tuple(factor_ids)

    def _access_control_facts(
        self,
        tenant: str,
        *,
        user_id: str | None,
        memberships: tuple[Membership, ...],
        active_context: ActiveAccessContext | None,
    ) -> tuple[AccessControlFact, ...]:
        facts: list[AccessControlFact] = []
        active_membership_ids = (
            set(active_context.membership_ids) if active_context is not None else set()
        )
        for membership in memberships:
            is_active = membership.membership_id in active_membership_ids
            facts.append(
                AccessControlFact(
                    tenant=tenant,
                    subject_type="group"
                    if membership.scope_type == AccessScopeType.GROUP.value
                    else "user",
                    subject_id=membership.scope_id
                    if membership.scope_type == AccessScopeType.GROUP.value
                    else membership.user_id,
                    user_id=membership.user_id,
                    scope_type=membership.scope_type,
                    scope_id=membership.scope_id,
                    role=membership.kind,
                    hat=active_context.hat if is_active and active_context else None,
                    access_profile_id=active_context.access_profile_id
                    if is_active and active_context
                    else None,
                    membership_id=membership.membership_id,
                    evidence_refs=self._access_fact_evidence_refs(
                        tenant, {membership.membership_id}
                    ),
                )
            )
        if active_context is not None:
            evidence_refs = self._access_fact_evidence_refs(
                tenant, {active_context.active_context_id}
            )
            facts.append(
                AccessControlFact(
                    tenant=tenant,
                    subject_type="user",
                    subject_id=active_context.user_id,
                    user_id=active_context.user_id,
                    scope_type=active_context.scope_type.value,
                    scope_id=active_context.scope_id,
                    role=active_context.hat,
                    hat=active_context.hat,
                    access_profile_id=active_context.access_profile_id,
                    evidence_refs=evidence_refs,
                )
            )
            for group_id in active_context.group_ids:
                facts.append(
                    AccessControlFact(
                        tenant=tenant,
                        subject_type="group",
                        subject_id=group_id,
                        user_id=active_context.user_id,
                        scope_type=active_context.scope_type.value,
                        scope_id=active_context.scope_id,
                        role=active_context.hat,
                        hat=active_context.hat,
                        access_profile_id=active_context.access_profile_id,
                        evidence_refs=evidence_refs,
                    )
                )
        return tuple(facts)

    def _access_fact_evidence_refs(
        self, tenant: str, subjects: set[str]
    ) -> tuple[CanonEntityReference, ...]:
        return tuple(
            CanonEntityReference(
                concept="Evidence Source",
                identifier=record.audit_id,
                scope=tenant,
                source_system="user-engine:audit",
                local_type=record.summary or record.action,
            )
            for record in self.store.audit_log()
            if record.tenant == tenant and record.subject in subjects
        )

    def _projection_access_context(
        self,
        user_id: str,
        tenant: str,
        *,
        application_id: str | None,
    ) -> Mapping[str, Any]:
        active_context = self.store.active_access_context(user_id, tenant)
        if active_context is None:
            return {}
        if (
            application_id is not None
            and active_context.service_id is not None
            and active_context.service_id != application_id
        ):
            return {}
        return {
            "active_hat": active_context.hat,
            "access_profile_id": active_context.access_profile_id,
            "scope_type": active_context.scope_type.value,
            "scope_id": active_context.scope_id,
            "realm_id": active_context.realm_id,
            "service_id": active_context.service_id,
            "asset_id": active_context.asset_id,
            "group_ids": active_context.group_ids,
            "claims": dict(active_context.claims),
            "profile_defaults": dict(active_context.profile_defaults),
            "factor_ids": active_context.factor_ids,
        }

    def _ensure_actor_session(
        self, actor: Actor, correlation_id: str
    ) -> UserSession:
        identity = self.store.find_identity(*actor.identity_key)
        if identity is not None:
            user = self._require_user(identity.user_id)
            account = self._require_account(user.user_id)
            return self._session(actor, user, account)

        decision = self._authorize(
            actor,
            action="me.read",
            resource_type="user-engine:me",
            resource_id=actor.subject,
            tenant=actor.tenant,
            correlation_id=correlation_id,
        )
        user = User(
            display_name=actor.preferred_username,
            primary_email=_optional_claim(actor, "email"),
        )
        account = Account(
            account_id=new_id("acct"),
            user_id=user.user_id,
            status=AccountStatus.ACTIVE,
        )
        tenant_account = TenantAccount(user_id=user.user_id, tenant=actor.tenant)
        external_identity = ExternalIdentity(
            identity_id=new_id("idn"),
            user_id=user.user_id,
            issuer=actor.issuer,
            subject=actor.subject,
            provider=actor.authorized_party,
        )
        with self.store.transaction():
            self.store.save_user(user)
            self.store.save_account(account)
            self.store.save_tenant_account(tenant_account)
            self.store.save_identity(external_identity)
            self._record_mutation(
                actor,
                action="user.create_from_identity",
                subject=user.user_id,
                tenant=actor.tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                event_type="user.created",
                aggregate_id=user.user_id,
                payload={
                    "user_id": user.user_id,
                    "account_id": account.account_id,
                    "identity": {"issuer": actor.issuer, "subject": actor.subject},
                },
            )
        return self._session(actor, user, account)

    def _ensure_family_dataspace_application(
        self,
        actor: Actor,
        request: FamilyDataspaceRequest,
        correlation_id: str,
    ) -> tuple[Application, ApplicationBinding]:
        desired_binding = _family_dataspace_binding(request)
        application = self.store.application(request.application_id)
        if application is not None:
            binding = self.store.binding(request.application_id)
            if binding is None:
                decision = self._authorize(
                    actor,
                    action="application.bind",
                    resource_type="user-engine:application",
                    resource_id=request.application_id,
                    tenant=request.tenant,
                    correlation_id=correlation_id,
                    application_id=request.application_id,
                )
                with self.store.transaction():
                    self.store.save_binding(desired_binding)
                    self._record_mutation(
                        actor,
                        action="application.bind",
                        subject=request.application_id,
                        tenant=request.tenant,
                        correlation_id=correlation_id,
                        decision_id=decision.decision_id,
                        event_type="application.bound",
                        aggregate_id=request.application_id,
                        payload={
                            "application_id": request.application_id,
                            "catalog_namespaces": desired_binding.catalog_namespaces,
                        },
                        application_id=request.application_id,
                    )
                binding = desired_binding
            return application, binding

        application = Application(
            application_id=request.application_id,
            display_name=request.application_display_name,
            owner=request.family_scope_id,
            allowed_profile_scopes=(
                ProfileScope.GLOBAL,
                ProfileScope.TENANT,
                ProfileScope.APPLICATION,
                ProfileScope.MEMBERSHIP,
            ),
            allowed_projection_types=(
                ProjectionType.APPLICATION_RUNTIME,
                ProjectionType.CLAIMS_ENRICHMENT,
            ),
        )
        return (
            self.register_application(
                actor,
                application,
                binding=desired_binding,
                correlation_id=correlation_id,
            ),
            desired_binding,
        )

    def _ensure_family_dataspace_catalog(
        self,
        actor: Actor,
        request: FamilyDataspaceRequest,
        correlation_id: str,
    ) -> Catalog:
        catalog_id = f"{request.application_id}.profile"
        existing = self.store.catalog(catalog_id)
        if existing is not None:
            return existing

        catalog = Catalog(
            catalog_id=catalog_id,
            namespace=request.catalog_namespace,
            version="0.1.0",
            owning_application_id=request.application_id,
            lifecycle=CatalogLifecycle.ACTIVE,
            attributes=(
                AttributeDefinition(
                    key=f"{request.catalog_namespace}.family_display_name",
                    value_type="string",
                    scope=ProfileScope.TENANT,
                    sensitivity=Sensitivity.PERSONAL,
                    visibility=(
                        Visibility.USER,
                        Visibility.ADMIN,
                        Visibility.APPLICATION,
                    ),
                    mutability=(Mutability.ADMIN,),
                    default=request.family_display_name,
                ),
                AttributeDefinition(
                    key=f"{request.catalog_namespace}.member_display_name",
                    value_type="string",
                    scope=ProfileScope.APPLICATION,
                    sensitivity=Sensitivity.PERSONAL,
                    visibility=(
                        Visibility.USER,
                        Visibility.ADMIN,
                        Visibility.APPLICATION,
                    ),
                    mutability=(Mutability.USER, Mutability.ADMIN),
                ),
                AttributeDefinition(
                    key=f"{request.catalog_namespace}.home_view",
                    value_type="string",
                    scope=ProfileScope.APPLICATION,
                    sensitivity=Sensitivity.INTERNAL,
                    visibility=(Visibility.USER, Visibility.APPLICATION),
                    mutability=(Mutability.USER,),
                    default="family",
                    validation={"enum": ["family", "personal"]},
                ),
            ),
        )
        return self.publish_catalog(
            actor,
            catalog,
            correlation_id=correlation_id,
        )

    def _ensure_membership(
        self,
        actor: Actor,
        user_id: str,
        *,
        tenant: str,
        scope_type: str,
        scope_id: str,
        kind: str,
        correlation_id: str,
    ) -> Membership:
        for membership in self.store.memberships_for_user(user_id, tenant=tenant):
            if (
                membership.scope_type == scope_type
                and membership.scope_id == scope_id
                and membership.kind == kind
            ):
                return membership
        return self.add_membership(
            actor,
            user_id,
            tenant=tenant,
            scope_type=scope_type,
            scope_id=scope_id,
            kind=kind,
            correlation_id=correlation_id,
        )

    def _apply_family_profile_defaults(
        self,
        actor: Actor,
        user_id: str,
        *,
        tenant: str,
        application_id: str,
        catalog_namespace: str,
        values: Mapping[str, Any],
        correlation_id: str,
    ) -> None:
        for key, value in values.items():
            self.set_profile_value(
                actor,
                user_id,
                _family_profile_key(catalog_namespace, key),
                value,
                scope=ProfileScope.APPLICATION,
                scope_id=application_id,
                tenant=tenant,
                application_id=application_id,
                correlation_id=correlation_id,
            )

    def _require_family_invitation(self, invitation_id: str) -> FamilyInvitation:
        invitation = self.store.family_invitation(invitation_id)
        if invitation is None:
            raise NotFoundError("family invitation not found")
        return invitation

    def _identity_entity_refs(
        self,
        actor: Actor,
        user: User,
        account: Account,
        identities: tuple[ExternalIdentity, ...],
        memberships: tuple[Membership, ...],
        tenant: str,
        application_id: str | None,
        active_access_context: ActiveAccessContext | None,
    ) -> Mapping[str, CanonEntityReference]:
        actor_identifier = f"{actor.issuer}:{actor.subject}"
        refs: dict[str, CanonEntityReference] = {
            "actor": CanonEntityReference(
                concept="Actor",
                identifier=actor_identifier,
                scope=actor.tenant,
                source_system=actor.issuer,
                local_type=actor.principal_type.value,
            ),
            "user": CanonEntityReference(
                concept="User",
                identifier=user.user_id,
                scope=tenant,
                local_type="user-engine-user-record",
            ),
            "account": CanonEntityReference(
                concept="Account",
                identifier=account.account_id,
                scope=tenant,
                local_type=account.status.value,
            ),
            "authenticated_subject": CanonEntityReference(
                concept="Authenticated Subject",
                identifier=actor_identifier,
                scope=actor.tenant,
                source_system=actor.issuer,
                local_type="issuer-subject",
            ),
            "authorization_principal": CanonEntityReference(
                concept="Authorization Principal",
                identifier=f"{tenant}:{actor_identifier}",
                scope=tenant,
                local_type=actor.principal_type.value,
            ),
            "tenant": CanonEntityReference(
                concept="Tenant",
                identifier=tenant,
                scope=tenant,
                local_type="isolation-scope",
            ),
        }
        if application_id is not None:
            refs["application_scope"] = CanonEntityReference(
                concept="Scope",
                identifier=application_id,
                scope=tenant,
                local_type="application",
            )
        if active_access_context is not None:
            refs["active_access_context"] = CanonEntityReference(
                concept="Active Access Context",
                identifier=active_access_context.active_context_id,
                scope=tenant,
                local_type=active_access_context.scope_type.value,
            )
            refs["active_hat"] = CanonEntityReference(
                concept="Hat",
                identifier=active_access_context.hat,
                scope=tenant,
                local_type="active",
            )
            refs["access_profile"] = CanonEntityReference(
                concept="Access Profile",
                identifier=active_access_context.access_profile_id,
                scope=tenant,
                local_type=active_access_context.hat,
            )
            if active_access_context.realm_id is not None:
                refs["realm"] = CanonEntityReference(
                    concept="Realm",
                    identifier=active_access_context.realm_id,
                    scope=tenant,
                    local_type="realm",
                )
            if active_access_context.service_id is not None:
                refs["service_area"] = CanonEntityReference(
                    concept="Service Area",
                    identifier=active_access_context.service_id,
                    scope=tenant,
                    local_type="service",
                )
            if active_access_context.asset_id is not None:
                refs["asset_scope"] = CanonEntityReference(
                    concept="Asset Scope",
                    identifier=active_access_context.asset_id,
                    scope=tenant,
                    local_type="asset",
                )
            for group_id in active_access_context.group_ids:
                refs[f"group:{group_id}"] = CanonEntityReference(
                    concept="Group",
                    identifier=group_id,
                    scope=tenant,
                    local_type="group",
                )
        for identity in identities:
            refs[f"identity_record:{identity.identity_id}"] = CanonEntityReference(
                concept="Identity Record",
                identifier=identity.identity_id,
                scope=identity.issuer,
                source_system=identity.issuer,
                local_type=identity.provider,
            )
            refs[f"identifier:{identity.identity_id}"] = CanonEntityReference(
                concept="Scoped Identifier",
                identifier=f"{identity.issuer}:{identity.subject}",
                scope=identity.issuer,
                source_system=identity.issuer,
                local_type="issuer-subject",
            )
        for membership in memberships:
            scope_key = f"{membership.scope_type}:{membership.scope_id}"
            refs.setdefault(
                scope_key,
                CanonEntityReference(
                    concept=_scope_concept(membership.scope_type),
                    identifier=membership.scope_id,
                    scope=tenant,
                    local_type=membership.scope_type,
                ),
            )
            refs[f"membership:{membership.membership_id}"] = CanonEntityReference(
                concept="Membership Relationship",
                identifier=membership.membership_id,
                scope=tenant,
                local_type=membership.kind,
            )
            refs[f"role:{membership.membership_id}"] = CanonEntityReference(
                concept="Role",
                identifier=f"{membership.scope_id}:{membership.kind}",
                scope=membership.scope_id,
                local_type=membership.kind,
            )
        return refs

    def _identity_relationship_refs(
        self,
        entity_refs: Mapping[str, CanonEntityReference],
        memberships: tuple[Membership, ...],
        tenant: str,
        evidence_id: str | None,
        *,
        active_access_context: ActiveAccessContext | None = None,
    ) -> tuple[tuple[CanonRelationshipReference, ...], tuple[CanonEntityReference, ...]]:
        account_ref = entity_refs["account"]
        subject_ref = entity_refs["authenticated_subject"]
        principal_ref = entity_refs["authorization_principal"]
        user_ref = entity_refs["user"]
        tenant_ref = entity_refs["tenant"]
        relationships = [
            CanonRelationshipReference(
                relationship_type="belongs_to_tenant",
                source=user_ref,
                target=tenant_ref,
                scope=tenant,
                evidence_id=evidence_id,
            ),
            CanonRelationshipReference(
                relationship_type="authenticates_as",
                source=account_ref,
                target=subject_ref,
                scope=tenant,
                evidence_id=evidence_id,
            ),
            CanonRelationshipReference(
                relationship_type="evaluated_as",
                source=subject_ref,
                target=principal_ref,
                scope=tenant,
                evidence_id=evidence_id,
            ),
        ]
        grant_like_refs: list[CanonEntityReference] = []
        for key, ref in entity_refs.items():
            if key.startswith("identity_record:"):
                relationships.append(
                    CanonRelationshipReference(
                        relationship_type="identity_link",
                        source=ref,
                        target=user_ref,
                        scope=ref.scope,
                        evidence_id=evidence_id,
                    )
                )
        for membership in memberships:
            membership_ref = entity_refs[f"membership:{membership.membership_id}"]
            role_ref = entity_refs[f"role:{membership.membership_id}"]
            scope_ref = entity_refs[f"{membership.scope_type}:{membership.scope_id}"]
            grant_ref = CanonEntityReference(
                concept="Access Grant",
                identifier=membership.membership_id,
                scope=tenant,
                local_type=membership.kind,
                evidence_id=evidence_id,
            )
            grant_like_refs.append(grant_ref)
            relationships.extend(
                (
                    CanonRelationshipReference(
                        relationship_type="member_of",
                        source=user_ref,
                        target=scope_ref,
                        scope=tenant,
                        evidence_id=evidence_id,
                    ),
                    CanonRelationshipReference(
                        relationship_type="role_label",
                        source=membership_ref,
                        target=role_ref,
                        scope=tenant,
                        evidence_id=evidence_id,
                    ),
                    CanonRelationshipReference(
                        relationship_type="scoped_to",
                        source=grant_ref,
                        target=tenant_ref,
                        scope=tenant,
                        evidence_id=evidence_id,
                    ),
                )
            )
        if active_access_context is not None:
            active_ref = entity_refs["active_access_context"]
            hat_ref = entity_refs["active_hat"]
            profile_ref = entity_refs["access_profile"]
            relationships.extend(
                (
                    CanonRelationshipReference(
                        relationship_type="wears_hat",
                        source=user_ref,
                        target=hat_ref,
                        scope=tenant,
                        evidence_id=evidence_id,
                    ),
                    CanonRelationshipReference(
                        relationship_type="selected_access_profile",
                        source=active_ref,
                        target=profile_ref,
                        scope=tenant,
                        evidence_id=evidence_id,
                    ),
                )
            )
        return tuple(relationships), tuple(grant_like_refs)

    def _identity_evidence_refs(
        self,
        user: User,
        account: Account,
        memberships: tuple[Membership, ...],
        tenant: str,
        *,
        active_access_context: ActiveAccessContext | None = None,
    ) -> tuple[CanonEntityReference, ...]:
        membership_ids = {membership.membership_id for membership in memberships}
        access_subjects = (
            {active_access_context.active_context_id}
            if active_access_context is not None
            else set()
        )
        subjects = {user.user_id, account.account_id, *membership_ids, *access_subjects}
        evidence = []
        for record in self.store.audit_log():
            if record.tenant != tenant:
                continue
            if record.subject not in subjects and record.summary != "membership.added":
                continue
            evidence.append(
                CanonEntityReference(
                    concept="Evidence Source",
                    identifier=record.audit_id,
                    scope=tenant,
                    source_system="user-engine:audit",
                    local_type=record.summary or record.action,
                )
            )
        return tuple(evidence)

    def _session(self, actor: Actor, user: User, account: Account) -> UserSession:
        identities = self.store.identities_for_user(user.user_id)
        return UserSession(actor=actor, user=user, account=account, identities=identities)

    def _authorize(
        self,
        actor: Actor,
        *,
        action: str,
        resource_type: str,
        resource_id: str,
        tenant: str,
        correlation_id: str,
        application_id: str | None = None,
        target_user_id: str | None = None,
        context: Mapping[str, Any] | None = None,
    ):
        request = AuthorizationRequest(
            actor=actor,
            resource_type=resource_type,
            resource_id=resource_id,
            action=action,
            tenant=tenant,
            correlation_id=correlation_id,
            application_id=application_id,
            target_user_id=target_user_id,
            context=self._authorization_context(
                actor,
                target_user_id=target_user_id,
                context=context,
                tenant=tenant,
            ),
        )
        decision = self.authorization.check(request)
        if decision.allowed:
            if decision.lifetime is None:
                self._record_authorization_denial(
                    actor,
                    action=action,
                    resource_id=resource_id,
                    tenant=tenant,
                    correlation_id=correlation_id,
                    application_id=application_id,
                    decision=AuthorizationDecision(
                        effect=decision.effect,
                        reason="authorization allow has no lifetime",
                    ),
                    summary="authorization denied (allow has no lifetime)",
                )
                raise AuthorizationDenied("authorization allow has no lifetime")
            if decision.expired():
                self._record_authorization_denial(
                    actor,
                    action=action,
                    resource_id=resource_id,
                    tenant=tenant,
                    correlation_id=correlation_id,
                    application_id=application_id,
                    decision=decision,
                    summary="authorization denied (allow expired)",
                )
                raise AuthorizationDenied("authorization allow expired")
            return decision
        summary = "authorization denied"
        if decision.stance_applied:
            summary = f"authorization denied (stance {decision.stance_applied})"
        self._record_authorization_denial(
            actor,
            action=action,
            resource_id=resource_id,
            tenant=tenant,
            correlation_id=correlation_id,
            application_id=application_id,
            decision=decision,
            summary=summary,
        )
        raise AuthorizationDenied(decision.reason or "authorization denied")

    def _record_authorization_denial(
        self,
        actor: Actor,
        *,
        action: str,
        resource_id: str,
        tenant: str,
        correlation_id: str,
        application_id: str | None,
        decision: AuthorizationDecision,
        summary: str,
    ) -> None:
        self.store.append_audit(
            AuditRecord(
                audit_id=new_id("aud"),
                actor=actor,
                action=action,
                subject=resource_id,
                tenant=tenant,
                correlation_id=correlation_id,
                decision_id=decision.decision_id,
                application_id=application_id,
                summary=summary,
            )
        )

    def _record_mutation(
        self,
        actor: Actor,
        *,
        action: str,
        subject: str,
        tenant: str,
        correlation_id: str,
        decision_id: str | None,
        event_type: str,
        aggregate_id: str,
        payload: Mapping[str, Any],
        application_id: str | None = None,
    ) -> None:
        self.store.append_audit(
            AuditRecord(
                audit_id=new_id("aud"),
                actor=actor,
                action=action,
                subject=subject,
                tenant=tenant,
                correlation_id=correlation_id,
                decision_id=decision_id,
                application_id=application_id,
                summary=event_type,
            )
        )
        self.store.append_outbox(
            OutboxEvent(
                event_id=new_id("evt"),
                event_type=event_type,
                aggregate_id=aggregate_id,
                payload=dict(payload),
                tenant=tenant,
                correlation_id=correlation_id,
            )
        )

    def _resolve_effective_profile(
        self, user_id: str, application_id: str | None, tenant: str | None
    ) -> EffectiveProfile:
        self._require_user(user_id)
        values: dict[str, Any] = {}
        source_layers: dict[str, str] = {}
        trace: dict[str, list[str]] = {}
        definitions = self._active_attribute_definitions()
        for key, definition in definitions.items():
            if definition.default is not None:
                values[key] = definition.default
                source_layers[key] = "catalog_default"
                trace[key] = ["catalog_default"]

        profile_values = self.store.values_for_user(user_id)
        for profile_value in profile_values:
            if profile_value.attribute_key not in definitions:
                continue
            if profile_value.scope != ProfileScope.GLOBAL:
                continue
            values[profile_value.attribute_key] = profile_value.value
            source_layers[profile_value.attribute_key] = "global"
            trace.setdefault(profile_value.attribute_key, []).append("global")

        for profile_value in profile_values:
            if profile_value.attribute_key not in definitions:
                continue
            if profile_value.scope != ProfileScope.TENANT:
                continue
            if profile_value.scope_id != tenant:
                continue
            values[profile_value.attribute_key] = profile_value.value
            layer = f"tenant:{tenant}"
            source_layers[profile_value.attribute_key] = layer
            trace.setdefault(profile_value.attribute_key, []).append(layer)

        for profile_value in profile_values:
            if profile_value.attribute_key not in definitions:
                continue
            if profile_value.scope != ProfileScope.APPLICATION:
                continue
            if profile_value.scope_id != application_id:
                continue
            values[profile_value.attribute_key] = profile_value.value
            layer = f"application:{application_id}"
            source_layers[profile_value.attribute_key] = layer
            trace.setdefault(profile_value.attribute_key, []).append(layer)

        return EffectiveProfile(
            user_id=user_id,
            application_id=application_id,
            values=values,
            source_layers=source_layers,
            trace={key: tuple(layers) for key, layers in trace.items()},
        )

    def _authorization_context(
        self,
        actor: Actor,
        *,
        target_user_id: str | None,
        context: Mapping[str, Any] | None,
        tenant: str,
    ) -> Mapping[str, Any]:
        facts: dict[str, Any] = {
            "actor_roles": actor.roles,
            "actor_scopes": actor.scopes,
            "assurance": dict(actor.assurance),
        }
        if target_user_id is not None:
            memberships = self.store.memberships_for_user(target_user_id, tenant=tenant)
            facts["target_memberships"] = tuple(
                {
                    "membership_id": membership.membership_id,
                    "scope_type": membership.scope_type,
                    "scope_id": membership.scope_id,
                    "kind": membership.kind,
                    "freshness_version": membership.freshness_version,
                }
                for membership in memberships
            )
            active_context = self.store.active_access_context(target_user_id, tenant)
            if active_context is not None:
                facts["active_access_context"] = {
                    "active_context_id": active_context.active_context_id,
                    "access_profile_id": active_context.access_profile_id,
                    "hat": active_context.hat,
                    "scope_type": active_context.scope_type.value,
                    "scope_id": active_context.scope_id,
                    "realm_id": active_context.realm_id,
                    "service_id": active_context.service_id,
                    "asset_id": active_context.asset_id,
                    "group_ids": active_context.group_ids,
                }
        if context:
            facts.update(context)
        return facts

    def _operation_tenant(
        self,
        actor: Actor,
        tenant: str | None,
        scope: ProfileScope,
        scope_id: str | None,
    ) -> str:
        if scope == ProfileScope.TENANT:
            return self.resolve_tenant_context(actor, scope_id).tenant
        return self.resolve_tenant_context(actor, tenant).tenant

    def _validate_catalog(self, catalog: Catalog) -> None:
        if self.store.application(catalog.owning_application_id) is None:
            raise ValidationError("catalog owning application is not registered")
        if catalog.lifecycle != CatalogLifecycle.ACTIVE:
            raise ValidationError("only active catalogs can be published")
        if not catalog.attributes:
            raise ValidationError("catalog must define at least one attribute")
        keys = [attribute.key for attribute in catalog.attributes]
        if len(keys) != len(set(keys)):
            raise ValidationError("catalog attribute keys must be unique")
        binding = self.store.binding(catalog.owning_application_id)
        if binding and catalog.namespace not in binding.catalog_namespaces:
            raise ValidationError("catalog namespace is not bound to application")

        existing_catalog = self.store.catalog(catalog.catalog_id)
        if existing_catalog is not None:
            self._validate_catalog_update(existing_catalog, catalog)

        active_definitions = self._active_attribute_definitions(
            excluding_catalog_id=catalog.catalog_id
        )
        active_catalogs = (
            existing
            for existing in self.store.all_catalogs()
            if existing.catalog_id != catalog.catalog_id
            and existing.lifecycle == CatalogLifecycle.ACTIVE
        )
        for existing in active_catalogs:
            if (
                existing.namespace == catalog.namespace
                and existing.owning_application_id != catalog.owning_application_id
            ):
                raise ConflictError("active catalog namespace is already owned")

        for attribute in catalog.attributes:
            if not attribute.key.startswith(f"{catalog.namespace}."):
                raise ValidationError("attribute keys must use the catalog namespace")
            if attribute.key in active_definitions:
                raise ConflictError("active catalog already owns attribute key")
            if not attribute.visibility:
                raise ValidationError("attribute must declare visibility")
            if not attribute.mutability:
                raise ValidationError("attribute must declare mutability")
            if Mutability.READ_ONLY in attribute.mutability and len(attribute.mutability) > 1:
                raise ValidationError("read-only mutability cannot be combined")
            if attribute.default is not None:
                self._validate_value(attribute, attribute.default)

    def _validate_catalog_update(self, previous: Catalog, catalog: Catalog) -> None:
        if previous.namespace != catalog.namespace:
            raise ValidationError("catalog namespace cannot change")
        if previous.owning_application_id != catalog.owning_application_id:
            raise ValidationError("catalog owner cannot change")
        if _version_tuple(catalog.version) < _version_tuple(previous.version):
            raise ValidationError("catalog version cannot move backwards")

        previous_attributes = {
            attribute.key: attribute for attribute in previous.attributes
        }
        for attribute in catalog.attributes:
            previous_attribute = previous_attributes.get(attribute.key)
            if previous_attribute is None:
                continue
            if _sensitivity_rank(attribute.sensitivity) < _sensitivity_rank(
                previous_attribute.sensitivity
            ):
                raise ValidationError("catalog update cannot downgrade sensitivity")

    def _validate_profile_scope(
        self,
        definition: AttributeDefinition,
        scope: ProfileScope,
        scope_id: str | None,
    ) -> None:
        allowed_scopes = {
            ProfileScope.GLOBAL,
            ProfileScope.TENANT,
            ProfileScope.APPLICATION,
            definition.scope,
        }
        if scope not in allowed_scopes:
            raise ValidationError("profile value scope is not allowed for attribute")
        if scope == ProfileScope.APPLICATION:
            if scope_id is None:
                raise ValidationError("application profile values require scope_id")
            if self.store.application(scope_id) is None:
                raise ValidationError("application profile scope_id is not registered")
        elif scope == ProfileScope.TENANT:
            if scope_id is None:
                raise ValidationError("tenant profile values require scope_id")
        elif scope_id is not None:
            raise ValidationError(
                "only tenant or application scoped values may set scope_id"
            )

    def _validate_value(self, definition: AttributeDefinition, value: Any) -> None:
        if value is None:
            return
        validators = {
            "string": lambda item: isinstance(item, str),
            "integer": lambda item: isinstance(item, int) and not isinstance(item, bool),
            "number": lambda item: isinstance(item, (int, float))
            and not isinstance(item, bool),
            "boolean": lambda item: isinstance(item, bool),
            "object": lambda item: isinstance(item, Mapping),
            "array": lambda item: isinstance(item, (list, tuple)),
        }
        validator = validators.get(definition.value_type)
        if validator is None:
            raise ValidationError(f"unsupported attribute value_type: {definition.value_type}")
        if not validator(value):
            raise ValidationError(f"{definition.key} must be {definition.value_type}")
        enum_values = definition.validation.get("enum")
        if enum_values is not None and value not in enum_values:
            raise ValidationError(f"{definition.key} is not an allowed value")

    def _require_user(self, user_id: str) -> User:
        user = self.store.user(user_id)
        if user is None:
            raise NotFoundError("user not found")
        return user

    def _require_account(self, user_id: str) -> Account:
        account = self.store.user_account(user_id)
        if account is None:
            raise NotFoundError("account not found")
        return account

    def _require_attribute(self, attribute_key: str) -> AttributeDefinition:
        definition = self._active_attribute_definitions().get(attribute_key)
        if definition is None:
            raise NotFoundError("active attribute definition not found")
        return definition

    def _active_attribute_definitions(
        self,
        *,
        excluding_catalog_id: str | None = None,
        owning_application_id: str | None = None,
    ) -> dict[str, AttributeDefinition]:
        definitions: dict[str, AttributeDefinition] = {}
        for catalog in self.store.all_catalogs():
            if catalog.catalog_id == excluding_catalog_id:
                continue
            if (
                owning_application_id is not None
                and catalog.owning_application_id != owning_application_id
            ):
                continue
            if catalog.lifecycle != CatalogLifecycle.ACTIVE:
                continue
            for attribute in catalog.attributes:
                definitions[attribute.key] = attribute
        return dict(sorted(definitions.items()))


def _optional_claim(actor: Actor, key: str) -> str | None:
    value = actor.claims.get(key)
    if value is None:
        return 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 _prepared_factor_signature(
    requirements: Iterable[PreparedFactorRequirement],
) -> tuple[tuple[str, str], ...]:
    return tuple(
        sorted(
            (
                requirement.factor_type.value,
                requirement.normalized_value.casefold(),
            )
            for requirement in requirements
        )
    )


def _prepared_account_expired(prepared: PreparedAccount) -> bool:
    return prepared.expires_at is not None and prepared.expires_at <= utc_now()


def _prepared_account_matches_factors(
    prepared: PreparedAccount, factors: Iterable[IdentityFactor]
) -> bool:
    now = utc_now()
    factor_values = {
        (factor.factor_type, factor.normalized_value.casefold())
        for factor in factors
        if factor.expires_at is None or factor.expires_at > now
    }
    return all(
        (
            requirement.factor_type,
            requirement.normalized_value.casefold(),
        )
        in factor_values
        for requirement in prepared.required_factor_matches
    )


def _access_profile_scope_id(profile: AccessProfile) -> str:
    if profile.scope_id:
        return profile.scope_id
    if profile.scope_type == AccessScopeType.TENANT:
        return profile.tenant
    raise ValidationError("access profile scope_id is required")


def _active_group_ids(
    profile: AccessProfile, memberships: Iterable[Membership]
) -> tuple[str, ...]:
    group_ids = {
        membership.scope_id
        for membership in memberships
        if membership.scope_type == AccessScopeType.GROUP.value
    }
    group_ids.update(profile.group_scope_ids)
    return tuple(sorted(group_ids))


def _onboarding_step_from_template(template: WelcomeProtocolStep) -> OnboardingStep:
    lifecycle_gap = (
        f"subsystem-callback-missing:{template.subsystem}:{template.step_key}"
        if template.requires_subsystem_callback and not template.callback_ref
        else None
    )
    initial_status = (
        OnboardingStepStatus.BLOCKED
        if lifecycle_gap is not None
        else OnboardingStepStatus.PENDING
    )
    task = OnboardingTask(
        subsystem=template.subsystem,
        task_kind=template.task_kind,
        external_ref=template.task_ref,
        status=initial_status,
    )
    handoff = SubsystemHandoff(
        subsystem=template.subsystem,
        callback_ref=template.callback_ref,
        lifecycle_gap=lifecycle_gap,
        status=initial_status,
    )
    return OnboardingStep(
        step_key=template.step_key,
        title=template.title,
        subsystem=template.subsystem,
        status=initial_status,
        task=task,
        handoff=handoff,
        support_ref=template.support_ref,
        lifecycle_gap=lifecycle_gap,
    )


def _activate_initial_onboarding_step(
    steps: tuple[OnboardingStep, ...]
) -> tuple[OnboardingStep, ...]:
    if any(step.status == OnboardingStepStatus.BLOCKED for step in steps):
        return steps
    for index, step in enumerate(steps):
        if step.status == OnboardingStepStatus.PENDING:
            return (
                *steps[:index],
                _onboarding_step_with_status(step, OnboardingStepStatus.IN_PROGRESS),
                *steps[index + 1 :],
            )
    return steps


def _replace_onboarding_step_status(
    steps: tuple[OnboardingStep, ...],
    step_key: str,
    *,
    status: OnboardingStepStatus,
    lifecycle_gap: str | None,
) -> tuple[OnboardingStep, ...]:
    updated = []
    found = False
    for step in steps:
        if step.step_key != step_key:
            updated.append(step)
            continue
        found = True
        updated.append(
            _onboarding_step_with_status(
                step,
                status,
                lifecycle_gap=lifecycle_gap,
            )
        )
    if not found:
        raise NotFoundError("onboarding step not found")
    return _activate_next_onboarding_step(tuple(updated))


def _activate_next_onboarding_step(
    steps: tuple[OnboardingStep, ...]
) -> tuple[OnboardingStep, ...]:
    if any(
        step.status
        in (OnboardingStepStatus.IN_PROGRESS, OnboardingStepStatus.BLOCKED)
        for step in steps
    ):
        return steps
    for index, step in enumerate(steps):
        if step.status == OnboardingStepStatus.PENDING:
            return (
                *steps[:index],
                _onboarding_step_with_status(step, OnboardingStepStatus.IN_PROGRESS),
                *steps[index + 1 :],
            )
    return steps


def _resume_onboarding_steps(
    steps: tuple[OnboardingStep, ...], callback_refs: Mapping[str, str]
) -> tuple[OnboardingStep, ...]:
    updated = []
    for step in steps:
        if step.status not in (
            OnboardingStepStatus.BLOCKED,
            OnboardingStepStatus.FAILED,
        ):
            updated.append(step)
            continue
        callback_ref = callback_refs.get(step.step_key)
        handoff = step.handoff
        if handoff is not None and callback_ref is not None:
            handoff = replace(
                handoff,
                callback_ref=callback_ref,
                lifecycle_gap=None,
                status=OnboardingStepStatus.IN_PROGRESS,
            )
        updated.append(
            _onboarding_step_with_status(
                step,
                OnboardingStepStatus.IN_PROGRESS,
                lifecycle_gap=None,
                handoff=handoff,
            )
        )
    return tuple(updated)


def _onboarding_step_with_status(
    step: OnboardingStep,
    status: OnboardingStepStatus,
    *,
    lifecycle_gap: str | None = None,
    handoff: SubsystemHandoff | None = None,
) -> OnboardingStep:
    task = replace(step.task, status=status) if step.task is not None else None
    resolved_handoff = handoff if handoff is not None else step.handoff
    if resolved_handoff is not None:
        resolved_handoff = replace(
            resolved_handoff,
            status=status,
            lifecycle_gap=lifecycle_gap,
        )
    return replace(
        step,
        status=status,
        task=task,
        handoff=resolved_handoff,
        lifecycle_gap=lifecycle_gap,
        updated_at=utc_now(),
    )


def _replace_onboarding_journey_steps(
    journey: OnboardingJourney, steps: tuple[OnboardingStep, ...]
) -> OnboardingJourney:
    now = utc_now()
    status = _journey_status_from_steps(steps)
    return replace(
        journey,
        steps=steps,
        status=status,
        active_step_key=_active_onboarding_step_key(steps),
        updated_at=now,
        completed_at=now if status == OnboardingJourneyStatus.COMPLETED else None,
        failed_at=now if status == OnboardingJourneyStatus.FAILED else journey.failed_at,
        skipped_at=now if status == OnboardingJourneyStatus.SKIPPED else None,
    )


def _journey_status_from_steps(
    steps: tuple[OnboardingStep, ...]
) -> OnboardingJourneyStatus:
    if any(step.status == OnboardingStepStatus.FAILED for step in steps):
        return OnboardingJourneyStatus.FAILED
    if any(step.status == OnboardingStepStatus.BLOCKED for step in steps):
        return OnboardingJourneyStatus.BLOCKED
    if all(step.status == OnboardingStepStatus.SKIPPED for step in steps):
        return OnboardingJourneyStatus.SKIPPED
    if all(
        step.status in (OnboardingStepStatus.COMPLETED, OnboardingStepStatus.SKIPPED)
        for step in steps
    ):
        return OnboardingJourneyStatus.COMPLETED
    if any(step.status == OnboardingStepStatus.IN_PROGRESS for step in steps):
        return OnboardingJourneyStatus.IN_PROGRESS
    return OnboardingJourneyStatus.PENDING


def _active_onboarding_step_key(steps: tuple[OnboardingStep, ...]) -> str | None:
    for status in (OnboardingStepStatus.IN_PROGRESS, OnboardingStepStatus.BLOCKED):
        for step in steps:
            if step.status == status:
                return step.step_key
    return None


def _scope_concept(scope_type: str) -> str:
    concepts = {
        "team": "Team",
        "tenant": "Tenant",
        "application": "Scope",
        "family": "Group",
        "group": "Group",
        "realm": "Realm",
        "service": "Service Area",
        "asset": "Asset Scope",
    }
    return concepts.get(scope_type, "Scope")


def _family_dataspace_binding(
    request: FamilyDataspaceRequest,
) -> ApplicationBinding:
    event_source = request.event_source or request.application_id
    return ApplicationBinding(
        application_id=request.application_id,
        oidc_client_id=request.oidc_client_id,
        protected_system_id=request.protected_system_id,
        catalog_namespaces=(request.catalog_namespace,),
        event_source=event_source,
        deployment_ref=request.deployment_ref,
    )


def _family_profile_key(catalog_namespace: str, key: str) -> str:
    if key.startswith(f"{catalog_namespace}."):
        return key
    return f"{catalog_namespace}.{key}"


def _family_role_value(role: FamilyRole | str) -> str:
    if str(role) in {"user", TENANT_ADMIN_ROLE}:
        return str(role)
    try:
        return FamilyRole(str(role)).value
    except ValueError as exc:
        raise ValidationError(f"unsupported family role: {role}") from exc


def _visible_in_projection(
    definition: AttributeDefinition, projection_type: ProjectionType
) -> bool:
    if projection_type == ProjectionType.AUDIT:
        return True
    visibility = {
        ProjectionType.SELF_SERVICE: Visibility.USER,
        ProjectionType.ADMIN: Visibility.ADMIN,
        ProjectionType.APPLICATION_RUNTIME: Visibility.APPLICATION,
        ProjectionType.AGENT_CONTEXT: Visibility.APPLICATION,
        ProjectionType.CLAIMS_ENRICHMENT: Visibility.APPLICATION,
    }.get(projection_type)
    if visibility is None:
        return False
    return visibility in definition.visibility


def _is_application_projection(projection_type: ProjectionType) -> bool:
    return projection_type in {
        ProjectionType.APPLICATION_RUNTIME,
        ProjectionType.AGENT_CONTEXT,
        ProjectionType.CLAIMS_ENRICHMENT,
    }


def _must_redact(
    definition: AttributeDefinition, projection_type: ProjectionType
) -> bool:
    if projection_type in {
        ProjectionType.ADMIN,
        ProjectionType.AUDIT,
        ProjectionType.SELF_SERVICE,
    }:
        return False
    return definition.sensitivity in {Sensitivity.SENSITIVE, Sensitivity.SECRET}


def _version_tuple(version: str) -> tuple[int, int, int]:
    parts = version.split(".")
    if len(parts) != 3 or any(not part.isdigit() for part in parts):
        raise ValidationError("catalog version must be semantic MAJOR.MINOR.PATCH")
    return tuple(int(part) for part in parts)


def _sensitivity_rank(sensitivity: Sensitivity) -> int:
    return {
        Sensitivity.PUBLIC: 0,
        Sensitivity.INTERNAL: 1,
        Sensitivity.PERSONAL: 2,
        Sensitivity.SENSITIVE: 3,
        Sensitivity.SECRET: 4,
    }[sensitivity]