feat: implement onboarding journeys

README.md

@@ -13,7 +13,8 @@ See `docs/development.md`, `docs/configuration.md`, `docs/contracts.md`,
 `docs/postgres-durable-store-consumer-requirements.md`, `docs/examples.md`,
 `docs/registration-identity-and-factor-model.md`,
 `docs/prepared-accounts-and-entitlement-claims.md`,
-`docs/hats-realms-services-assets-access-profiles.md`, `docs/scenarios.md`,
+`docs/hats-realms-services-assets-access-profiles.md`,
+`docs/onboarding-journeys-and-welcome-protocols.md`, `docs/scenarios.md`,
 `docs/operability.md`, `docs/release.md`, `docs/ui-contracts.md`,
 `docs/identity-domain-naming-decision.md`, and `docs/final-assessment.md`
 for implementation boundaries, contracts, canon mappings, examples, and release

SCOPE.md

@@ -62,6 +62,6 @@ through `USER-WP-0015`. `USER-WP-0010` implements the first headless
 registration and factor-evidence slice. `USER-WP-0011` implements prepared
 accounts and entitlement claims. `USER-WP-0012` implements hats, realms,
 services, assets, access profiles, active context, and exportable
-access-control facts. `USER-WP-0013` through `USER-WP-0015` remain proposed
-future workplans for onboarding journeys, optional UI, and security
-conformance.
+access-control facts. `USER-WP-0013` implements onboarding journeys and
+welcome protocols. `USER-WP-0014` and `USER-WP-0015` remain proposed future
+workplans for optional UI and security conformance.

docs/contracts.md

@@ -14,6 +14,12 @@ HTTP or RPC adapters should preserve these operation names:
   `claim_prepared_account`
 - `register_access_profile`, `list_access_profiles`, `select_active_hat`,
   `export_access_control_facts`, `access_profile_diagnostics`
+- `register_welcome_protocol`, `list_welcome_protocols`,
+  `start_onboarding_journey`, `start_onboarding_for_registration`,
+  `start_onboarding_for_prepared_account`, `progress_onboarding_step`,
+  `complete_onboarding_step`, `skip_onboarding_step`,
+  `fail_onboarding_step`, `resume_onboarding_journey`,
+  `onboarding_diagnostics`
 - `me`, `create_user`, `set_account_status`, `link_identity`
 - `resolve_tenant_context`, `set_tenant_account_status`, `add_membership`,
   `tenant_diagnostics`
@@ -86,14 +92,33 @@ Access-profile diagnostics report counts, factor requirement types, and
 approval-required issues without exposing profile default values, projection
 claim values, or raw factor values.
 
+## Onboarding Journey Contract
+
+Welcome protocols are tenant-scoped onboarding templates. They can match
+registration completion, prepared-account claims, invitations, access-profile
+events, or manual starts by trigger type and optional context keys.
+
+Onboarding journeys are persisted user state. They track protocol, source
+event, trigger type, ordered steps, task references, subsystem handoff
+references, lifecycle gaps, active step, status, and correlation ids.
+
+Registration completion and prepared-account claim automatically start matching
+welcome protocols. Manual start/progress/complete/skip/fail/resume operations
+are also exposed through `UserEngineService` and authorization-gated.
+
+Missing required subsystem callbacks produce explicit lifecycle gaps and block
+the journey. The service records audit/outbox events with ids, statuses, step
+keys, source ids, and lifecycle gap identifiers, but not factor values, support
+content, notification payloads, or subsystem-specific tour data.
+
 ## Identity Context Contract
 
 `identity_context` is the first canon-facing read model for NetKingdom
 identity-domain consumers. It resolves a verified actor into the local user,
 account, external identity links, tenant scope, memberships, optional
 application scope, optional effective profile, optional active access context,
-exportable access-control facts, canon entity references, relationship
-references, grant-like membership facts, and evidence references.
+exportable access-control facts, onboarding journeys, canon entity references,
+relationship references, grant-like membership facts, and evidence references.
 
 The method keeps these concepts distinct:
 

docs/netkingdom-registration-onboarding-vision.md

@@ -235,9 +235,9 @@ once.
 
 ## Recommended Workplans
 
-As of 2026-06-15, `USER-WP-0010`, `USER-WP-0011`, and `USER-WP-0012` are
-implemented as headless user-engine slices. The later workplans remain
-recommended follow-on work.
+As of 2026-06-15, `USER-WP-0010`, `USER-WP-0011`, `USER-WP-0012`, and
+`USER-WP-0013` are implemented as headless user-engine slices. The later
+workplans remain recommended follow-on work.
 
 | Workplan | Title | Purpose |
 | --- | --- | --- |

docs/onboarding-journeys-and-welcome-protocols.md

@@ -0,0 +1,91 @@
+# Onboarding Journeys And Welcome Protocols
+
+Status: implemented headless slice
+Date: 2026-06-15
+Related workplan: USER-WP-0013
+
+## Purpose
+
+This slice adds resumable onboarding journeys for newly registered or newly
+entitled users. Welcome protocols are templates that can be triggered by
+registration completion, prepared-account claims, invitations, access-profile
+events, or explicit manual starts.
+
+user-engine owns journey state, audit/event correlation, lifecycle gaps, and
+adapter references. Notification delivery, support content, protected service
+tours, and external workflow/task systems remain outside the core domain.
+
+## Domain Model
+
+`WelcomeProtocol` is the tenant-scoped template. It stores trigger type,
+optional matching keys such as prepared journey, application, realm, service,
+role, hat, factor requirements, and ordered `WelcomeProtocolStep` definitions.
+
+`WelcomeProtocolStep` names the subsystem, task kind, optional external task
+reference, optional callback reference, support reference, and whether a
+subsystem callback is required.
+
+`OnboardingJourney` is the persisted user state. It records tenant, user,
+protocol, trigger type, source id/event, journey key, status, active step,
+correlation id, and timestamps.
+
+`OnboardingStep` carries step status plus optional `OnboardingTask` and
+`SubsystemHandoff` references. Missing required subsystem callbacks produce
+explicit lifecycle gaps such as
+`subsystem-callback-missing:<subsystem>:<step>`.
+
+## Public Facade
+
+`UserEngineService` exposes:
+
+- `register_welcome_protocol(...)`
+- `list_welcome_protocols(...)`
+- `start_onboarding_journey(...)`
+- `start_onboarding_for_registration(...)`
+- `start_onboarding_for_prepared_account(...)`
+- `progress_onboarding_step(...)`
+- `complete_onboarding_step(...)`
+- `skip_onboarding_step(...)`
+- `fail_onboarding_step(...)`
+- `resume_onboarding_journey(...)`
+- `onboarding_diagnostics(...)`
+
+Registration completion auto-starts matching registration protocols. Prepared
+account claim auto-starts matching prepared-account protocols when the claimed
+package includes a matching onboarding journey key.
+
+## Lifecycle Rules
+
+Journey status is derived from step state:
+
+- a missing required callback starts the journey as `blocked`;
+- an active non-blocked first step starts as `in_progress`;
+- failed steps fail the journey;
+- all completed/skipped steps complete the journey;
+- all skipped steps skip the journey.
+
+Resuming a pending, blocked, or failed journey can attach callback references
+and returns blocked/failed steps to `in_progress`.
+
+## Adapter Boundary
+
+The service defines ports for notification delivery, task systems, support
+content, subsystem welcome callbacks, and lifecycle task linking. The current
+headless slice stores adapter references but does not call external systems.
+
+## Identity Context And Diagnostics
+
+`identity_context` includes onboarding journeys for the resolved user/tenant.
+`onboarding_diagnostics` reports protocol count, journey count, status counts,
+blocked step ids, and lifecycle gaps.
+
+Diagnostics and outbox events avoid factor values and service content. Payloads
+carry ids, statuses, trigger types, source ids, active step keys, and lifecycle
+gap identifiers.
+
+## Current Limits
+
+- No notification platform or support-content renderer is implemented.
+- No protected subsystem tour is hard-coded into user-engine.
+- External task and callback execution is left to adapters.
+- UI surfaces are left to USER-WP-0014.

docs/prepared-accounts-and-entitlement-claims.md

@@ -114,4 +114,5 @@ addresses, and eID payloads.
   adds explicit approval decisions.
 - Final authorization policy and ACL evaluation remains outside user-engine;
   user-engine only activates owned facts for policy systems to consume.
-- Journey orchestration beyond outbox requests is left to USER-WP-0013.
+- Journey orchestration from prepared-account onboarding requests is
+  implemented by USER-WP-0013.

docs/registration-identity-and-factor-model.md

@@ -113,6 +113,7 @@ return factor values.
 - Hats, realms, services, assets, and access profiles are implemented by
   USER-WP-0012 and documented in
   `docs/hats-realms-services-assets-access-profiles.md`.
-- Welcome protocols and onboarding journeys are left to USER-WP-0013.
+- Welcome protocols and onboarding journeys are implemented by USER-WP-0013
+  and documented in `docs/onboarding-journeys-and-welcome-protocols.md`.
 - Registration UI is left to USER-WP-0014.
 - Provider-backed proofing and credential flows remain external adapters.

src/user_engine/adapters/local.py

@@ -22,6 +22,7 @@ from user_engine.domain import (
     FamilyInvitation,
     IdentityFactor,
     Membership,
+    OnboardingJourney,
     OutboxEvent,
     PreparedAccount,
     ProfileScope,
@@ -29,6 +30,7 @@ from user_engine.domain import (
     RegistrationSession,
     TenantAccount,
     User,
+    WelcomeProtocol,
 )
 
 SCHEMA_VERSION = "0001_initial"
@@ -62,6 +64,8 @@ class InMemoryUserEngineStore:
     active_access_contexts: dict[
         tuple[str, str], ActiveAccessContext
     ] = field(default_factory=dict)
+    welcome_protocols: dict[str, WelcomeProtocol] = field(default_factory=dict)
+    onboarding_journeys: dict[str, OnboardingJourney] = field(default_factory=dict)
     profile_values: dict[
         tuple[str, str, ProfileScope, str | None], ProfileValue
     ] = field(default_factory=dict)
@@ -234,6 +238,45 @@ class InMemoryUserEngineStore:
             if context.tenant == tenant
         )
 
+    def save_welcome_protocol(self, protocol: WelcomeProtocol) -> None:
+        self.welcome_protocols[protocol.protocol_id] = protocol
+
+    def welcome_protocol(self, protocol_id: str) -> WelcomeProtocol | None:
+        return self.welcome_protocols.get(protocol_id)
+
+    def welcome_protocols_for_tenant(
+        self, tenant: str
+    ) -> tuple[WelcomeProtocol, ...]:
+        return tuple(
+            protocol
+            for protocol in self.welcome_protocols.values()
+            if protocol.tenant == tenant
+        )
+
+    def save_onboarding_journey(self, journey: OnboardingJourney) -> None:
+        self.onboarding_journeys[journey.journey_id] = journey
+
+    def onboarding_journey(self, journey_id: str) -> OnboardingJourney | None:
+        return self.onboarding_journeys.get(journey_id)
+
+    def onboarding_journeys_for_user(
+        self, user_id: str, *, tenant: str | None = None
+    ) -> tuple[OnboardingJourney, ...]:
+        return tuple(
+            journey
+            for journey in self.onboarding_journeys.values()
+            if journey.user_id == user_id and (tenant is None or journey.tenant == tenant)
+        )
+
+    def onboarding_journeys_for_tenant(
+        self, tenant: str
+    ) -> tuple[OnboardingJourney, ...]:
+        return tuple(
+            journey
+            for journey in self.onboarding_journeys.values()
+            if journey.tenant == tenant
+        )
+
     def save_profile_value(self, value: ProfileValue) -> None:
         self.profile_values[
             (value.user_id, value.attribute_key, value.scope, value.scope_id)
@@ -296,6 +339,8 @@ class InMemoryUserEngineStore:
             "prepared_accounts": len(self.prepared_accounts),
             "access_profiles": len(self.access_profiles),
             "active_access_contexts": len(self.active_access_contexts),
+            "welcome_protocols": len(self.welcome_protocols),
+            "onboarding_journeys": len(self.onboarding_journeys),
             "profile_values": len(self.profile_values),
             "audit_records": len(self.audit_records),
             "pending_outbox_events": len(self.outbox_events),
@@ -317,6 +362,8 @@ class InMemoryUserEngineStore:
             "prepared_accounts": copy.deepcopy(self.prepared_accounts),
             "access_profiles": copy.deepcopy(self.access_profiles),
             "active_access_contexts": copy.deepcopy(self.active_access_contexts),
+            "welcome_protocols": copy.deepcopy(self.welcome_protocols),
+            "onboarding_journeys": copy.deepcopy(self.onboarding_journeys),
             "profile_values": copy.deepcopy(self.profile_values),
             "audit_records": copy.deepcopy(self.audit_records),
             "outbox_events": copy.deepcopy(self.outbox_events),
@@ -349,6 +396,10 @@ class InMemoryUserEngineStore:
         self.active_access_contexts = snapshot[
             "active_access_contexts"
         ]  # type: ignore[assignment]
+        self.welcome_protocols = snapshot["welcome_protocols"]  # type: ignore[assignment]
+        self.onboarding_journeys = snapshot[
+            "onboarding_journeys"
+        ]  # type: ignore[assignment]
         self.profile_values = snapshot["profile_values"]  # type: ignore[assignment]
         self.audit_records = [*snapshot_audit_records, *denied_audit_records]
         self.outbox_events = snapshot["outbox_events"]  # type: ignore[assignment]

src/user_engine/domain/__init__.py

@@ -32,6 +32,12 @@ from user_engine.domain.models import (
     ManagementMode,
     Membership,
     Mutability,
+    OnboardingJourney,
+    OnboardingJourneyStatus,
+    OnboardingStep,
+    OnboardingStepStatus,
+    OnboardingTask,
+    OnboardingTriggerType,
     OutboxEvent,
     PrincipalType,
     PreparedAccount,
@@ -45,9 +51,12 @@ from user_engine.domain.models import (
     RegistrationSession,
     RegistrationStatus,
     Sensitivity,
+    SubsystemHandoff,
     TenantAccount,
     User,
     Visibility,
+    WelcomeProtocol,
+    WelcomeProtocolStep,
     new_id,
     utc_now,
 )
@@ -84,6 +93,12 @@ __all__ = [
     "ManagementMode",
     "Membership",
     "Mutability",
+    "OnboardingJourney",
+    "OnboardingJourneyStatus",
+    "OnboardingStep",
+    "OnboardingStepStatus",
+    "OnboardingTask",
+    "OnboardingTriggerType",
     "OutboxEvent",
     "PrincipalType",
     "PreparedAccount",
@@ -97,9 +112,12 @@ __all__ = [
     "RegistrationSession",
     "RegistrationStatus",
     "Sensitivity",
+    "SubsystemHandoff",
     "TenantAccount",
     "User",
     "Visibility",
+    "WelcomeProtocol",
+    "WelcomeProtocolStep",
     "new_id",
     "utc_now",
 ]

src/user_engine/domain/models.py

@@ -101,6 +101,32 @@ class AccessScopeType(StrEnum):
     GROUP = "group"
 
 
+class OnboardingTriggerType(StrEnum):
+    REGISTRATION = "registration"
+    PREPARED_ACCOUNT = "prepared_account"
+    INVITATION = "invitation"
+    ACCESS_PROFILE = "access_profile"
+    MANUAL = "manual"
+
+
+class OnboardingJourneyStatus(StrEnum):
+    PENDING = "pending"
+    IN_PROGRESS = "in_progress"
+    BLOCKED = "blocked"
+    COMPLETED = "completed"
+    SKIPPED = "skipped"
+    FAILED = "failed"
+
+
+class OnboardingStepStatus(StrEnum):
+    PENDING = "pending"
+    IN_PROGRESS = "in_progress"
+    BLOCKED = "blocked"
+    COMPLETED = "completed"
+    SKIPPED = "skipped"
+    FAILED = "failed"
+
+
 class ProfileScope(StrEnum):
     GLOBAL = "global"
     TENANT = "tenant"
@@ -375,6 +401,91 @@ class AccessControlFact:
     source_system: str = "user-engine"
 
 
+@dataclass(frozen=True)
+class OnboardingTask:
+    subsystem: str
+    task_kind: str
+    task_id: str = field(default_factory=lambda: new_id("otask"))
+    external_ref: str | None = None
+    status: OnboardingStepStatus = OnboardingStepStatus.PENDING
+
+
+@dataclass(frozen=True)
+class SubsystemHandoff:
+    subsystem: str
+    handoff_id: str = field(default_factory=lambda: new_id("hnd"))
+    callback_ref: str | None = None
+    lifecycle_gap: str | None = None
+    status: OnboardingStepStatus = OnboardingStepStatus.PENDING
+
+
+@dataclass(frozen=True)
+class WelcomeProtocolStep:
+    step_key: str
+    title: str
+    subsystem: str
+    task_kind: str = "welcome"
+    task_ref: str | None = None
+    callback_ref: str | None = None
+    support_ref: str | None = None
+    requires_subsystem_callback: bool = False
+
+
+@dataclass(frozen=True)
+class WelcomeProtocol:
+    tenant: str
+    name: str
+    trigger_type: OnboardingTriggerType
+    steps: tuple[WelcomeProtocolStep, ...]
+    protocol_id: str = field(default_factory=lambda: new_id("wpro"))
+    journey_key: str | None = None
+    registration_status: RegistrationStatus | None = None
+    prepared_journey: str | None = None
+    application_id: str | None = None
+    realm_id: str | None = None
+    service_id: str | None = None
+    role: str | None = None
+    hat: str | None = None
+    required_factor_types: tuple[IdentityFactorType, ...] = ()
+    created_at: datetime = field(default_factory=utc_now)
+    updated_at: datetime = field(default_factory=utc_now)
+
+
+@dataclass(frozen=True)
+class OnboardingStep:
+    step_key: str
+    title: str
+    subsystem: str
+    step_id: str = field(default_factory=lambda: new_id("ostep"))
+    status: OnboardingStepStatus = OnboardingStepStatus.PENDING
+    task: OnboardingTask | None = None
+    handoff: SubsystemHandoff | None = None
+    support_ref: str | None = None
+    lifecycle_gap: str | None = None
+    updated_at: datetime = field(default_factory=utc_now)
+
+
+@dataclass(frozen=True)
+class OnboardingJourney:
+    tenant: str
+    user_id: str
+    protocol_id: str
+    trigger_type: OnboardingTriggerType
+    steps: tuple[OnboardingStep, ...]
+    journey_id: str = field(default_factory=lambda: new_id("ojrn"))
+    status: OnboardingJourneyStatus = OnboardingJourneyStatus.PENDING
+    source_id: str | None = None
+    source_event_type: str | None = None
+    journey_key: str | None = None
+    active_step_key: str | None = None
+    correlation_id: str | None = None
+    created_at: datetime = field(default_factory=utc_now)
+    updated_at: datetime = field(default_factory=utc_now)
+    completed_at: datetime | None = None
+    failed_at: datetime | None = None
+    skipped_at: datetime | None = None
+
+
 @dataclass(frozen=True)
 class FamilyMemberSpec:
     primary_email: str

src/user_engine/ports.py

@@ -28,12 +28,14 @@ from user_engine.domain import (
     FamilyInvitation,
     IdentityFactor,
     Membership,
+    OnboardingJourney,
     OutboxEvent,
     PreparedAccount,
     ProfileValue,
     RegistrationSession,
     TenantAccount,
     User,
+    WelcomeProtocol,
 )
 
 
@@ -185,6 +187,33 @@ class UserEngineStore(Protocol):
     ) -> tuple[ActiveAccessContext, ...]:
         """Return active access contexts for a tenant."""
 
+    def save_welcome_protocol(self, protocol: WelcomeProtocol) -> None:
+        """Create or replace a welcome protocol template."""
+
+    def welcome_protocol(self, protocol_id: str) -> WelcomeProtocol | None:
+        """Return a welcome protocol template by id."""
+
+    def welcome_protocols_for_tenant(
+        self, tenant: str
+    ) -> tuple[WelcomeProtocol, ...]:
+        """Return welcome protocol templates for a tenant."""
+
+    def save_onboarding_journey(self, journey: OnboardingJourney) -> None:
+        """Create or replace an onboarding journey."""
+
+    def onboarding_journey(self, journey_id: str) -> OnboardingJourney | None:
+        """Return an onboarding journey by id."""
+
+    def onboarding_journeys_for_user(
+        self, user_id: str, *, tenant: str | None = None
+    ) -> tuple[OnboardingJourney, ...]:
+        """Return onboarding journeys for a user."""
+
+    def onboarding_journeys_for_tenant(
+        self, tenant: str
+    ) -> tuple[OnboardingJourney, ...]:
+        """Return onboarding journeys for a tenant."""
+
     def save_profile_value(self, value: ProfileValue) -> None:
         """Create or replace a profile value."""
 
@@ -260,6 +289,41 @@ class AccessControlFactExporter(Protocol):
         """Return an adapter-neutral access-control fact manifest."""
 
 
+class OnboardingNotificationPort(Protocol):
+    """Notify a delivery system about onboarding journey state."""
+
+    def notify(self, journey: OnboardingJourney) -> Mapping[str, Any]:
+        """Return adapter metadata for a notification request."""
+
+
+class OnboardingTaskPort(Protocol):
+    """Create or link external lifecycle tasks for onboarding steps."""
+
+    def link_task(self, journey: OnboardingJourney, step_key: str) -> Mapping[str, Any]:
+        """Return task-link metadata for one onboarding step."""
+
+
+class SupportContentPort(Protocol):
+    """Resolve support or help content references for onboarding."""
+
+    def content_ref(self, protocol: WelcomeProtocol, step_key: str) -> str | None:
+        """Return an adapter-owned content reference for a protocol step."""
+
+
+class SubsystemWelcomePort(Protocol):
+    """Call a protected subsystem welcome callback."""
+
+    def start(self, journey: OnboardingJourney, step_key: str) -> Mapping[str, Any]:
+        """Return callback metadata for a subsystem welcome step."""
+
+
+class LifecycleTaskLinkPort(Protocol):
+    """Link onboarding journeys to external lifecycle task systems."""
+
+    def link(self, journey: OnboardingJourney) -> Mapping[str, Any]:
+        """Return lifecycle task references for an onboarding journey."""
+
+
 class EventOutbox(Protocol):
     """Persist and publish durable domain events."""
 

src/user_engine/service.py

@@ -35,6 +35,12 @@ from user_engine.domain import (
     InvitationStatus,
     Membership,
     Mutability,
+    OnboardingJourney,
+    OnboardingJourneyStatus,
+    OnboardingStep,
+    OnboardingStepStatus,
+    OnboardingTask,
+    OnboardingTriggerType,
     OutboxEvent,
     PreparedAccount,
     PreparedAccountStatus,
@@ -47,9 +53,12 @@ from user_engine.domain import (
     RegistrationSession,
     RegistrationStatus,
     Sensitivity,
+    SubsystemHandoff,
     TenantAccount,
     User,
     Visibility,
+    WelcomeProtocol,
+    WelcomeProtocolStep,
     new_id,
     utc_now,
 )
@@ -142,6 +151,7 @@ class IdentityContext:
     gaps: tuple[str, ...] = ()
     active_access_context: ActiveAccessContext | None = None
     access_control_facts: tuple[AccessControlFact, ...] = ()
+    onboarding_journeys: tuple[OnboardingJourney, ...] = ()
 
 
 @dataclass(frozen=True)
@@ -167,6 +177,22 @@ class AccessProfileDiagnostics:
     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
@@ -549,6 +575,16 @@ class UserEngineService:
                     ),
                 },
             )
+            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,
@@ -979,6 +1015,16 @@ class UserEngineService:
                         "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,
@@ -1254,6 +1300,350 @@ class UserEngineService:
             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],
@@ -1803,6 +2193,9 @@ class UserEngineService:
             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,
@@ -1844,6 +2237,7 @@ class UserEngineService:
             gaps=gaps,
             active_access_context=active_access_context,
             access_control_facts=access_control_facts,
+            onboarding_journeys=onboarding_journeys,
         )
 
     def onboard_family_dataspace(
@@ -2702,6 +3096,241 @@ class UserEngineService:
             )
         )
 
+    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:
@@ -3869,6 +4498,201 @@ def _active_group_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",

tests/test_onboarding_journeys.py

@@ -0,0 +1,361 @@
+import unittest
+
+from user_engine.adapters.local import InMemoryUserEngineStore, LocalAuthorizationCheckPort
+from user_engine.domain import (
+    FactorVerification,
+    IdentityFactorType,
+    OnboardingJourneyStatus,
+    OnboardingStepStatus,
+    OnboardingTriggerType,
+    PreparedEntitlement,
+    PreparedEntitlementKind,
+    PreparedFactorRequirement,
+    WelcomeProtocol,
+    WelcomeProtocolStep,
+)
+from user_engine.service import UserEngineService
+from user_engine.testing.fixtures import FixtureIdentityClaimsAdapter, human_actor_claims
+
+
+class OnboardingJourneyTests(unittest.TestCase):
+    def test_registration_completion_starts_matching_welcome_protocol(self):
+        service, store = _service()
+        actor = _actor()
+        protocol = service.register_welcome_protocol(
+            actor,
+            _registration_protocol(),
+            correlation_id="corr-register-protocol",
+        )
+
+        completion = _complete_registration(service, actor)
+
+        journeys = store.onboarding_journeys_for_user(completion.user.user_id)
+        context = service.identity_context(
+            actor,
+            user_id=completion.user.user_id,
+            tenant="tenant:coulomb",
+            correlation_id="corr-context",
+        )
+
+        self.assertEqual(len(journeys), 1)
+        self.assertEqual(journeys[0].protocol_id, protocol.protocol_id)
+        self.assertEqual(journeys[0].trigger_type, OnboardingTriggerType.REGISTRATION)
+        self.assertEqual(journeys[0].status, OnboardingJourneyStatus.IN_PROGRESS)
+        self.assertEqual(journeys[0].steps[0].status, OnboardingStepStatus.IN_PROGRESS)
+        self.assertEqual(context.onboarding_journeys[0].journey_id, journeys[0].journey_id)
+        self.assertIn(
+            "onboarding_journey.started",
+            [event.event_type for event in service.outbox_events()],
+        )
+        self.assertNotIn(
+            "sample.user@example.test",
+            repr([event.payload for event in service.outbox_events()]),
+        )
+
+    def test_prepared_account_claim_starts_prepared_welcome_protocol(self):
+        service, store = _service()
+        actor = _actor()
+        service.register_welcome_protocol(
+            actor,
+            _prepared_protocol(),
+            correlation_id="corr-prepared-protocol",
+        )
+        prepared = service.prepare_account(
+            actor,
+            tenant="tenant:coulomb",
+            required_factor_matches=(
+                PreparedFactorRequirement(
+                    factor_type=IdentityFactorType.EMAIL,
+                    normalized_value="sample.user@example.test",
+                ),
+            ),
+            entitlements=(
+                PreparedEntitlement(
+                    kind=PreparedEntitlementKind.ONBOARDING_JOURNEY,
+                    tenant="tenant:coulomb",
+                    onboarding_journey="welcome-demo",
+                ),
+            ),
+            correlation_id="corr-prepare",
+        )
+        completion = _complete_registration(service, actor)
+
+        service.claim_prepared_account(
+            actor,
+            completion.session.registration_id,
+            prepared_account_id=prepared.prepared_account_id,
+            correlation_id="corr-claim",
+        )
+
+        journeys = store.onboarding_journeys_for_user(completion.user.user_id)
+        self.assertEqual(len(journeys), 1)
+        self.assertEqual(journeys[0].trigger_type, OnboardingTriggerType.PREPARED_ACCOUNT)
+        self.assertEqual(journeys[0].source_id, prepared.prepared_account_id)
+        self.assertEqual(journeys[0].journey_key, "welcome-demo")
+        self.assertIn(
+            "prepared_account.onboarding_requested",
+            [event.event_type for event in service.outbox_events()],
+        )
+        self.assertIn(
+            "onboarding_journey.started",
+            [event.event_type for event in service.outbox_events()],
+        )
+
+    def test_missing_subsystem_callback_blocks_journey_with_gap(self):
+        service, store = _service()
+        session = service.me(human_actor_claims(), correlation_id="corr-me")
+        protocol = service.register_welcome_protocol(
+            session.actor,
+            WelcomeProtocol(
+                tenant="tenant:coulomb",
+                name="Blocked Welcome",
+                trigger_type=OnboardingTriggerType.MANUAL,
+                steps=(
+                    WelcomeProtocolStep(
+                        step_key="external-setup",
+                        title="External Setup",
+                        subsystem="ops-console",
+                        requires_subsystem_callback=True,
+                    ),
+                ),
+            ),
+            correlation_id="corr-blocked-protocol",
+        )
+
+        start = service.start_onboarding_journey(
+            session.actor,
+            session.user.user_id,
+            protocol.protocol_id,
+            correlation_id="corr-start-blocked",
+        )
+        diagnostics = service.onboarding_diagnostics(
+            session.actor,
+            tenant="tenant:coulomb",
+            correlation_id="corr-diagnostics",
+        )
+
+        self.assertEqual(start.journey.status, OnboardingJourneyStatus.BLOCKED)
+        self.assertEqual(start.journey.steps[0].status, OnboardingStepStatus.BLOCKED)
+        self.assertIn("subsystem-callback-missing", start.journey.steps[0].lifecycle_gap)
+        self.assertEqual(store.record_counts()["onboarding_journeys"], 1)
+        self.assertEqual(diagnostics.statuses[OnboardingJourneyStatus.BLOCKED.value], 1)
+        self.assertTrue(diagnostics.lifecycle_gaps)
+
+    def test_blocked_journey_can_resume_and_complete(self):
+        service, _ = _service()
+        session = service.me(human_actor_claims(), correlation_id="corr-me")
+        protocol = service.register_welcome_protocol(
+            session.actor,
+            WelcomeProtocol(
+                tenant="tenant:coulomb",
+                name="Resume Welcome",
+                trigger_type=OnboardingTriggerType.MANUAL,
+                steps=(
+                    WelcomeProtocolStep(
+                        step_key="callback",
+                        title="Callback",
+                        subsystem="crm",
+                        requires_subsystem_callback=True,
+                    ),
+                ),
+            ),
+            correlation_id="corr-resume-protocol",
+        )
+        blocked = service.start_onboarding_journey(
+            session.actor,
+            session.user.user_id,
+            protocol.protocol_id,
+            correlation_id="corr-start",
+        ).journey
+
+        resumed = service.resume_onboarding_journey(
+            session.actor,
+            blocked.journey_id,
+            callback_refs={"callback": "crm://welcome/callback"},
+            correlation_id="corr-resume",
+        )
+        completed = service.complete_onboarding_step(
+            session.actor,
+            resumed.journey_id,
+            "callback",
+            correlation_id="corr-complete-step",
+        )
+
+        self.assertEqual(resumed.status, OnboardingJourneyStatus.IN_PROGRESS)
+        self.assertEqual(
+            resumed.steps[0].handoff.callback_ref,
+            "crm://welcome/callback",
+        )
+        self.assertIsNone(resumed.steps[0].lifecycle_gap)
+        self.assertEqual(completed.status, OnboardingJourneyStatus.COMPLETED)
+        self.assertEqual(completed.completed_at is not None, True)
+
+    def test_progress_skip_and_fail_operations_are_auditable(self):
+        service, _ = _service()
+        session = service.me(human_actor_claims(), correlation_id="corr-me")
+        protocol = service.register_welcome_protocol(
+            session.actor,
+            WelcomeProtocol(
+                tenant="tenant:coulomb",
+                name="Two Step Welcome",
+                trigger_type=OnboardingTriggerType.MANUAL,
+                steps=(
+                    WelcomeProtocolStep(
+                        step_key="intro",
+                        title="Intro",
+                        subsystem="portal",
+                    ),
+                    WelcomeProtocolStep(
+                        step_key="tour",
+                        title="Tour",
+                        subsystem="portal",
+                    ),
+                ),
+            ),
+            correlation_id="corr-two-step-protocol",
+        )
+        journey = service.start_onboarding_journey(
+            session.actor,
+            session.user.user_id,
+            protocol.protocol_id,
+            correlation_id="corr-start-two-step",
+        ).journey
+
+        progressed = service.progress_onboarding_step(
+            session.actor,
+            journey.journey_id,
+            "intro",
+            correlation_id="corr-progress",
+        )
+        second_active = service.complete_onboarding_step(
+            session.actor,
+            progressed.journey_id,
+            "intro",
+            correlation_id="corr-complete-intro",
+        )
+        skipped = service.skip_onboarding_step(
+            session.actor,
+            second_active.journey_id,
+            "tour",
+            correlation_id="corr-skip-tour",
+        )
+
+        self.assertEqual(skipped.status, OnboardingJourneyStatus.COMPLETED)
+        self.assertIn(
+            "onboarding_step.skipped",
+            [event.event_type for event in service.outbox_events()],
+        )
+
+        failed_protocol = service.register_welcome_protocol(
+            session.actor,
+            WelcomeProtocol(
+                tenant="tenant:coulomb",
+                name="Failing Welcome",
+                trigger_type=OnboardingTriggerType.MANUAL,
+                steps=(
+                    WelcomeProtocolStep(
+                        step_key="danger",
+                        title="Danger",
+                        subsystem="portal",
+                    ),
+                ),
+            ),
+            correlation_id="corr-fail-protocol",
+        )
+        failed_start = service.start_onboarding_journey(
+            session.actor,
+            session.user.user_id,
+            failed_protocol.protocol_id,
+            correlation_id="corr-start-fail",
+        ).journey
+        failed = service.fail_onboarding_step(
+            session.actor,
+            failed_start.journey_id,
+            "danger",
+            lifecycle_gap="portal-unavailable",
+            correlation_id="corr-fail-step",
+        )
+
+        self.assertEqual(failed.status, OnboardingJourneyStatus.FAILED)
+        self.assertEqual(failed.steps[0].lifecycle_gap, "portal-unavailable")
+        self.assertIn(
+            "onboarding_step.failed",
+            [event.event_type for event in service.outbox_events()],
+        )
+
+
+def _service():
+    store = InMemoryUserEngineStore()
+    service = UserEngineService(
+        store=store,
+        identity_adapter=FixtureIdentityClaimsAdapter(),
+        authorization=LocalAuthorizationCheckPort(),
+    )
+    return service, store
+
+
+def _actor():
+    return FixtureIdentityClaimsAdapter().normalize(
+        human_actor_claims(subject="sample-user", tenant="tenant:coulomb")
+    )
+
+
+def _complete_registration(service: UserEngineService, actor):
+    session = service.start_registration(actor, correlation_id="corr-start")
+    service.attach_registration_factor(
+        actor,
+        session.registration_id,
+        FactorVerification(
+            factor_type=IdentityFactorType.EMAIL,
+            normalized_value="sample.user@example.test",
+            display_value="sample.user@example.test",
+            source_system="fixture-email",
+        ),
+        correlation_id="corr-factor",
+    )
+    return service.complete_registration(
+        actor,
+        session.registration_id,
+        correlation_id="corr-complete",
+    )
+
+
+def _registration_protocol() -> WelcomeProtocol:
+    return WelcomeProtocol(
+        tenant="tenant:coulomb",
+        name="Registration Welcome",
+        trigger_type=OnboardingTriggerType.REGISTRATION,
+        required_factor_types=(IdentityFactorType.EMAIL,),
+        steps=(
+            WelcomeProtocolStep(
+                step_key="intro",
+                title="Intro",
+                subsystem="portal",
+                callback_ref="portal://welcome/intro",
+                requires_subsystem_callback=True,
+            ),
+        ),
+    )
+
+
+def _prepared_protocol() -> WelcomeProtocol:
+    return WelcomeProtocol(
+        tenant="tenant:coulomb",
+        name="Prepared Welcome",
+        trigger_type=OnboardingTriggerType.PREPARED_ACCOUNT,
+        journey_key="welcome-demo",
+        prepared_journey="welcome-demo",
+        steps=(
+            WelcomeProtocolStep(
+                step_key="prepared-intro",
+                title="Prepared Intro",
+                subsystem="portal",
+                callback_ref="portal://welcome/prepared",
+                requires_subsystem_callback=True,
+            ),
+        ),
+    )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_ports_and_fixtures.py

@@ -164,12 +164,14 @@ class _ProtocolOnlyStore:
         "identities",
         "identity_factors",
         "memberships",
+        "onboarding_journeys",
         "outbox_events",
         "prepared_accounts",
         "profile_values",
         "registration_sessions",
         "tenant_accounts",
         "users",
+        "welcome_protocols",
     }
 
     def __init__(self, inner: InMemoryUserEngineStore) -> None:

workplans/USER-WP-0013-onboarding-journeys-and-welcome-protocols.md

@@ -4,7 +4,7 @@ type: workplan
 title: "Onboarding Journeys And Welcome Protocols"
 domain: netkingdom
 repo: user-engine
-status: proposed
+status: finished
 owner: codex
 topic_slug: netkingdom
 planning_priority: medium
@@ -42,7 +42,7 @@ channels, and external task systems remain adapters or downstream systems.
 
 ```task
 id: USER-WP-0013-T1
-status: todo
+status: done
 priority: high
 state_hub_task_id: "30ef8507-eebc-4b96-8aa6-c530bef05739"
 ```
@@ -52,7 +52,7 @@ handoff models.
 
 ```task
 id: USER-WP-0013-T2
-status: todo
+status: done
 priority: high
 state_hub_task_id: "7c6e53d4-ff96-4036-a413-f04b4b73d266"
 ```
@@ -62,7 +62,7 @@ tenant, realm, service, application, role, hat, and factor requirements.
 
 ```task
 id: USER-WP-0013-T3
-status: todo
+status: done
 priority: high
 state_hub_task_id: "d9c2983a-45d1-4b1b-a416-63e180ca74b3"
 ```
@@ -72,7 +72,7 @@ with authorization, audit, and outbox behavior.
 
 ```task
 id: USER-WP-0013-T4
-status: todo
+status: done
 priority: medium
 state_hub_task_id: "7155c2eb-4e32-46f0-ad33-961784cb9a03"
 ```
@@ -82,7 +82,7 @@ welcome callbacks, and lifecycle task linking.
 
 ```task
 id: USER-WP-0013-T5
-status: todo
+status: done
 priority: medium
 state_hub_task_id: "c5e42dd6-207a-4b1e-a0d8-35701e9f71bc"
 ```
@@ -105,3 +105,41 @@ contracts.
 - Welcome protocol service facade.
 - Adapter ports for notifications and subsystem handoff.
 - Scenario tests for successful, blocked, and resumed onboarding.
+
+## Implementation Notes
+
+Implemented on 2026-06-15:
+
+- Added `OnboardingTriggerType`, `OnboardingJourneyStatus`,
+  `OnboardingStepStatus`, `WelcomeProtocol`, `WelcomeProtocolStep`,
+  `OnboardingJourney`, `OnboardingStep`, `OnboardingTask`, and
+  `SubsystemHandoff` domain models.
+- Added welcome-protocol and onboarding-journey persistence to
+  `UserEngineStore` and `InMemoryUserEngineStore`, including transaction
+  snapshots and record counts.
+- Added adapter ports for onboarding notifications, task links, support
+  content, subsystem welcome callbacks, and lifecycle task linking.
+- Added `UserEngineService` onboarding facade methods:
+  `register_welcome_protocol`, `list_welcome_protocols`,
+  `start_onboarding_journey`, `start_onboarding_for_registration`,
+  `start_onboarding_for_prepared_account`, `progress_onboarding_step`,
+  `complete_onboarding_step`, `skip_onboarding_step`,
+  `fail_onboarding_step`, `resume_onboarding_journey`, and
+  `onboarding_diagnostics`.
+- Added auto-start hooks for matching registration-completion protocols and
+  prepared-account claim protocols.
+- Extended `identity_context` with onboarding journeys for the resolved
+  user/tenant.
+- Added lifecycle-gap handling for missing subsystem callbacks and resumable
+  blocked/failed journey state.
+- Added `docs/onboarding-journeys-and-welcome-protocols.md`, public contract
+  updates, and tests for registration-triggered, prepared-claim-triggered,
+  blocked, resumed, progressed, skipped, and failed onboarding.
+
+Verification:
+
+```text
+make test
+Ran 66 tests in 0.620s
+OK
+```