diff --git a/api/controllers/console/notification.py b/api/controllers/console/notification.py
index 3e58f598bf7..080080bb361 100644
--- a/api/controllers/console/notification.py
+++ b/api/controllers/console/notification.py
@@ -1,56 +1,16 @@
-from collections.abc import Mapping
-from typing import TypedDict
-
 from flask_restx import Resource
 from pydantic import BaseModel, Field
 
 from controllers.common.fields import SimpleResultResponse
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.console import console_ns
-from controllers.console.wraps import (
-    account_initialization_required,
-    model_validate,
-    only_edition_cloud,
-    setup_required,
-    with_current_user,
-)
+from controllers.console.flask_admission import console_account_admission
+from controllers.console.wraps import model_validate
+from enums import DeploymentEdition
+from extensions.ext_application_services import application_services
 from fields.base import ResponseModel
-from libs.login import login_required
-from models import Account
-from services.billing_service import BillingService
-
-# Notification content is stored under three lang tags.
-_FALLBACK_LANG = "en-US"
-
-
-class NotificationLangContent(TypedDict, total=False):
-    lang: str
-    title: str
-    subtitle: str
-    body: str
-    titlePicUrl: str
-
-
-class NotificationItemDict(TypedDict):
-    notification_id: str | None
-    frequency: str | None
-    lang: str
-    title: str
-    subtitle: str
-    body: str
-    title_pic_url: str
-
-
-class NotificationResponseDict(TypedDict):
-    should_show: bool
-    notifications: list[NotificationItemDict]
-
-
-def _pick_lang_content(contents: Mapping[str, NotificationLangContent], lang: str) -> NotificationLangContent:
-    """Return the single LangContent for *lang*, falling back to English."""
-    return (
-        contents.get(lang) or contents.get(_FALLBACK_LANG) or next(iter(contents.values()), NotificationLangContent())
-    )
+from libs.helper import dump_response
+from machinery.context import RequestContext
 
 
 class DismissNotificationPayload(BaseModel):
@@ -92,39 +52,10 @@ class NotificationApi(Resource):
         },
     )
     @console_ns.response(200, "Success", console_ns.models[NotificationResponse.__name__])
-    @setup_required
-    @login_required
-    @with_current_user
-    @account_initialization_required
-    @only_edition_cloud
-    def get(self, current_user: Account):
-        result = BillingService.get_account_notification(str(current_user.id))
-
-        # Proto JSON uses camelCase field names (Kratos default marshaling).
-        response: NotificationResponseDict
-        if not result.get("shouldShow"):
-            response = {"should_show": False, "notifications": []}
-            return response, 200
-
-        lang = current_user.interface_language or _FALLBACK_LANG
-
-        notifications: list[NotificationItemDict] = []
-        for notification in result.get("notifications") or []:
-            contents: Mapping[str, NotificationLangContent] = notification.get("contents") or {}
-            lang_content = _pick_lang_content(contents, lang)
-            item: NotificationItemDict = {
-                "notification_id": notification.get("notificationId"),
-                "frequency": notification.get("frequency"),
-                "lang": lang_content.get("lang", lang),
-                "title": lang_content.get("title", ""),
-                "subtitle": lang_content.get("subtitle", ""),
-                "body": lang_content.get("body", ""),
-                "title_pic_url": lang_content.get("titlePicUrl", ""),
-            }
-            notifications.append(item)
-
-        response = {"should_show": bool(notifications), "notifications": notifications}
-        return response, 200
+    @console_account_admission(editions=frozenset({DeploymentEdition.CLOUD}))
+    def get(self, request_context: RequestContext):
+        result = application_services().notifications.get_active(request_context)
+        return dump_response(NotificationResponse, result), 200
 
 
 @console_ns.route("/notification/dismiss")
@@ -134,17 +65,10 @@ class NotificationDismissApi(Resource):
         description="Mark a notification as dismissed for the current user.",
         responses={200: "Success", 401: "Unauthorized"},
     )
-    @setup_required
-    @login_required
-    @with_current_user
-    @account_initialization_required
-    @only_edition_cloud
+    @console_account_admission(editions=frozenset({DeploymentEdition.CLOUD}))
     @console_ns.expect(console_ns.models[DismissNotificationPayload.__name__])
     @console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
     @model_validate(DismissNotificationPayload)
-    def post(self, payload: DismissNotificationPayload, current_user: Account):
-        BillingService.dismiss_notification(
-            notification_id=payload.notification_id,
-            account_id=str(current_user.id),
-        )
-        return {"result": "success"}, 200
+    def post(self, payload: DismissNotificationPayload, request_context: RequestContext):
+        application_services().notifications.dismiss(request_context, payload.notification_id)
+        return dump_response(SimpleResultResponse, {"result": "success"}), 200
diff --git a/api/controllers/console/onboarding.py b/api/controllers/console/onboarding.py
index f26e2d539e4..cbd77752e7b 100644
--- a/api/controllers/console/onboarding.py
+++ b/api/controllers/console/onboarding.py
@@ -7,36 +7,20 @@ action-based so callers do not replace server-side arrays with stale snapshots.
 """
 
 from datetime import datetime
-from typing import Literal, cast
 
 from flask_restx import Resource
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 from controllers.common.schema import register_response_schema_models, register_schema_models
-from extensions.ext_database import db
+from controllers.console.flask_admission import console_account_admission
+from controllers.console.wraps import model_validate
+from extensions.ext_application_services import application_services
 from fields.base import ResponseModel
 from libs.helper import dump_response
-from libs.login import login_required
-from models import Account
-from services.step_by_step_tour_service import StepByStepTourPatch, StepByStepTourService
+from machinery.context import RequestContext
+from services.entities.onboarding_entities import StepByStepTourAction, StepByStepTourPatch, StepByStepTourTaskId
 
 from . import console_ns
-from .wraps import (
-    account_initialization_required,
-    model_validate,
-    setup_required,
-    with_current_tenant_id,
-    with_current_user,
-)
-
-StepByStepTourAction = Literal[
-    "skip",
-    "complete_task",
-    "uncomplete_task",
-    "enable_current_workspace",
-    "disable_current_workspace",
-]
-StepByStepTourTaskId = Literal["home", "studio", "knowledge", "integration"]
 
 
 class StepByStepTourStatePatchPayload(BaseModel):
@@ -74,39 +58,22 @@ class StepByStepTourStateApi(Resource):
     @console_ns.doc("get_step_by_step_tour_state")
     @console_ns.doc(description="Get account-level Step-by-step Tour state")
     @console_ns.response(200, "Success", console_ns.models[StepByStepTourStateResponse.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @with_current_user
-    @with_current_tenant_id
-    def get(self, current_tenant_id: str, current_user: Account):
+    @console_account_admission()
+    def get(self, request_context: RequestContext):
         return dump_response(
             StepByStepTourStateResponse,
-            StepByStepTourService.get_state(
-                account=current_user,
-                current_tenant_id=current_tenant_id,
-                session=db.session,
-            ),
+            application_services().step_by_step_tour.get_state(request_context),
         )
 
     @console_ns.doc("patch_step_by_step_tour_state")
     @console_ns.doc(description="Update account-level Step-by-step Tour state")
     @console_ns.expect(console_ns.models[StepByStepTourStatePatchPayload.__name__])
     @console_ns.response(200, "Success", console_ns.models[StepByStepTourStateResponse.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @with_current_user
-    @with_current_tenant_id
+    @console_account_admission()
     @model_validate(StepByStepTourStatePatchPayload)
-    def patch(self, req_data: StepByStepTourStatePatchPayload, current_tenant_id: str, current_user: Account):
-        patch = cast(StepByStepTourPatch, req_data.model_dump(exclude_unset=True, exclude_none=True))
+    def patch(self, req_data: StepByStepTourStatePatchPayload, request_context: RequestContext):
+        patch = StepByStepTourPatch(action=req_data.action, task_id=req_data.task_id)
         return dump_response(
             StepByStepTourStateResponse,
-            StepByStepTourService.patch_state(
-                account=current_user,
-                current_tenant_id=current_tenant_id,
-                patch=patch,
-                session=db.session,
-            ),
+            application_services().step_by_step_tour.patch_state(request_context, patch),
         )
diff --git a/api/dev/generate_swagger_markdown_docs.py b/api/dev/generate_swagger_markdown_docs.py
index 991a487c107..a9451c52778 100644
--- a/api/dev/generate_swagger_markdown_docs.py
+++ b/api/dev/generate_swagger_markdown_docs.py
@@ -76,6 +76,10 @@ def _schema_markdown_type(schema: object) -> str:
         item_type = _schema_markdown_type(schema.get("items"))
         return f"[ {item_type or 'object'} ]"
     if isinstance(schema_type, str):
+        enum_values = schema.get("enum")
+        if isinstance(enum_values, list) and enum_values:
+            rendered_values = ", ".join(json.dumps(value, ensure_ascii=False) for value in enum_values)
+            return f"{schema_type}, <br>**Available values:** {rendered_values}"
         return schema_type
 
     return ""
diff --git a/api/extensions/ext_application_services.py b/api/extensions/ext_application_services.py
index 7a482248cce..747c658a561 100644
--- a/api/extensions/ext_application_services.py
+++ b/api/extensions/ext_application_services.py
@@ -31,6 +31,7 @@ from repositories.factory import DifyAPIRepositoryFactory
 from repositories.installation_state_repository import InstallationStateRepository
 from repositories.oauth_server_repository import RedisOAuthServerTokenRepository, SQLAlchemyOAuthServerRepository
 from repositories.recommended_app_catalog_repository import DatabaseRecommendedAppCatalogRepository
+from repositories.step_by_step_tour_repository import SQLAlchemyStepByStepTourStateRepository
 from repositories.tag_repository import TagRepository
 from repositories.trial_app_query_repository import TrialAppQueryRepository
 from repositories.trial_app_usage_repository import TrialAppUsageRepository
@@ -104,6 +105,8 @@ from services.feature_service import FeatureService
 from services.feature_service_gateway import FeatureServiceGateway
 from services.file_service import FileService
 from services.init_validation_service import InitValidationService
+from services.notification_gateway import BillingNotificationGateway
+from services.notification_service import NotificationService
 from services.notion_data_source_gateway import NotionDataSourceGateway
 from services.oauth_server_service import OAUTH_ACCESS_TOKEN_EXPIRES_IN, OAuthServerService
 from services.partner_tenant_binding_service import PartnerTenantBindingService
@@ -122,6 +125,7 @@ from services.retention.workflow_run.archive_log_service import WorkflowRunArchi
 from services.schema_definition_service import SchemaDefinitionService
 from services.setup_adapters import RedisSetupLock, RegisterServiceAccountProvisioner
 from services.setup_service import SetupService
+from services.step_by_step_tour_service import StepByStepTourService
 from services.tag_application_service import TagApplicationService
 from services.trial_app_usage import TrialAppUsageRecorder
 from services.web_app_runtime_query_service import WebAppRuntimeQueryService
@@ -188,6 +192,8 @@ class ApplicationServices:
     feature_queries: FeatureQueryService
     oauth_server: OAuthServerService
     init_validation: InitValidationService
+    notifications: NotificationService
+    step_by_step_tour: StepByStepTourService
     partner_tenant_bindings: PartnerTenantBindingService
     recommended_app_queries: RecommendedAppQueryService
     trial_app_usage: TrialAppUsageRecorder
@@ -434,6 +440,16 @@ def build_application_services(
             validation_required=(deployment_edition != DeploymentEdition.CLOUD and bool(initialization_password)),
             expected_password=initialization_password,
         ),
+        notifications=NotificationService(
+            accounts=accounts,
+            notifications=BillingNotificationGateway(),
+        ),
+        step_by_step_tour=StepByStepTourService(
+            accounts=accounts,
+            states=SQLAlchemyStepByStepTourStateRepository(session_factory=database_client),
+            enabled=dify_config.ENABLE_STEP_BY_STEP_TOUR,
+            rollout_started_at=dify_config.STEP_BY_STEP_TOUR_ROLLOUT_STARTED_AT,
+        ),
         partner_tenant_bindings=PartnerTenantBindingService(
             sync_bindings=BillingService.sync_partner_tenants_bindings,
         ),
diff --git a/api/openapi/markdown/console-openapi.md b/api/openapi/markdown/console-openapi.md
index b8008497e43..c143fd19243 100644
--- a/api/openapi/markdown/console-openapi.md
+++ b/api/openapi/markdown/console-openapi.md
@@ -13501,7 +13501,7 @@ default (the config form sends the full desired feature state on save).
 | mode | string, <br>**Available values:** "advanced-chat", "agent", "agent-chat", "all", "channel", "chat", "completion", "workflow", <br>**Default:** all | App mode filter<br>*Enum:* `"advanced-chat"`, `"agent"`, `"agent-chat"`, `"all"`, `"channel"`, `"chat"`, `"completion"`, `"workflow"` | No |
 | name | string | Filter by app name | No |
 | page | integer, <br>**Default:** 1 | Page number (1-99999) | No |
-| publication_status | string | Filter by published or draft Agent configuration status | No |
+| publication_status | string, <br>**Available values:** "drafts", "published" | Filter by published or draft Agent configuration status | No |
 | sort_by | string, <br>**Available values:** "earliest_created", "last_modified", "recently_created", <br>**Default:** last_modified | Sort apps by last modified, recently created, or earliest created<br>*Enum:* `"earliest_created"`, `"last_modified"`, `"recently_created"` | No |
 | tag_ids | [ string ] | Filter by tag IDs | No |
 
@@ -15744,7 +15744,7 @@ AppMCPServer Status Enum
 | copyright | string |  | No |
 | custom_disclaimer | string |  | No |
 | customize_domain | string |  | No |
-| customize_token_strategy | string |  | No |
+| customize_token_strategy | string, <br>**Available values:** "allow", "must", "not_allow" |  | No |
 | default_language | string |  | No |
 | description | string |  | No |
 | icon | string |  | No |
@@ -16202,7 +16202,7 @@ TEAM: Team collaboration paid plan
 | files | [ object ] |  | No |
 | inputs | object |  | Yes |
 | query | string |  | No |
-| response_mode | string |  | No |
+| response_mode | string, <br>**Available values:** "blocking", "streaming" |  | No |
 | retriever_from | string, <br>**Default:** explore_app |  | No |
 
 #### CompletionMessagePayload
@@ -16223,7 +16223,7 @@ TEAM: Team collaboration paid plan
 | files | [ object ] |  | No |
 | inputs | object |  | Yes |
 | query | string |  | No |
-| response_mode | string |  | No |
+| response_mode | string, <br>**Available values:** "blocking", "streaming" |  | No |
 | retriever_from | string, <br>**Default:** explore_app |  | No |
 
 #### ComplianceDownloadQuery
@@ -18263,9 +18263,9 @@ Flask blueprint initialization.
 | ---- | ---- | ----------- | -------- |
 | end_date | string | End date (YYYY-MM-DD) | No |
 | format | string, <br>**Available values:** "csv", "json", <br>**Default:** csv | Export format<br>*Enum:* `"csv"`, `"json"` | No |
-| from_source | string | Filter by feedback source | No |
+| from_source | string, <br>**Available values:** "admin", "user" | Filter by feedback source | No |
 | has_comment | boolean | Only include feedback with comments | No |
-| rating | string | Filter by rating | No |
+| rating | string, <br>**Available values:** "dislike", "like" | Filter by rating | No |
 | start_date | string | Start date (YYYY-MM-DD) | No |
 
 #### FeedbackStat
@@ -18663,7 +18663,7 @@ Icon information model.
 | ---- | ---- | ----------- | -------- |
 | icon | string |  | No |
 | icon_background | string |  | No |
-| icon_type | string |  | No |
+| icon_type | string, <br>**Available values:** "emoji", "image" |  | No |
 | icon_url | string |  | No |
 
 #### IconType
@@ -19245,7 +19245,7 @@ Enum class for large language model mode.
 | ---- | ---- | ----------- | -------- |
 | content | string | Optional text feedback providing additional detail. | No |
 | message_id | string | Message ID | Yes |
-| rating | string | Feedback rating. Set to `null` to revoke previously submitted feedback. | No |
+| rating | string, <br>**Available values:** "dislike", "like" | Feedback rating. Set to `null` to revoke previously submitted feedback. | No |
 
 #### MessageFile
 
@@ -19306,7 +19306,7 @@ Metadata Filtering Condition.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | conditions | [ [Condition](#condition) ] | List of metadata conditions to evaluate. | No |
-| logical_operator | string | How to combine multiple conditions. | No |
+| logical_operator | string, <br>**Available values:** "and", "or" | How to combine multiple conditions. | No |
 
 #### MetadataOperationData
 
@@ -19439,7 +19439,7 @@ Enum class for model property key.
 | is_exhausted | boolean |  | Yes |
 | is_unlimited | boolean |  | Yes |
 | next_credit_reset_date | integer |  | Yes |
-| pool_type | string |  | Yes |
+| pool_type | string, <br>**Available values:** "paid", "trial" |  | Yes |
 | quota_limit | integer | Credit limit for the effective pool; -1 means unlimited. | Yes |
 | quota_used | integer |  | Yes |
 | remaining_credits | integer | Remaining credits; -1 means unlimited. | Yes |
@@ -21434,7 +21434,7 @@ Model class for provider quota configuration.
 | ---- | ---- | ----------- | -------- |
 | metadata_filtering_conditions | [MetadataFilteringCondition](#metadatafilteringcondition) | Restrict retrieval to chunks whose document metadata matches the given conditions. Conditions are evaluated server-side against document metadata fields. | No |
 | reranking_enable | boolean | Whether reranking is enabled. | Yes |
-| reranking_mode | string | Reranking mode. Required when `reranking_enable` is `true`. | No |
+| reranking_mode | string, <br>**Available values:** "reranking_model", "weighted_score" | Reranking mode. Required when `reranking_enable` is `true`. | No |
 | reranking_model | [RerankingModel](#rerankingmodel) | Reranking model configuration. | No |
 | score_threshold | number | Minimum similarity score for results. Only effective when score threshold filtering is enabled. | No |
 | score_threshold_enabled | boolean | Whether score threshold filtering is enabled. | Yes |
@@ -21488,7 +21488,7 @@ Model class for provider quota configuration.
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| parent_mode | string | Parent-child segmentation mode. | No |
+| parent_mode | string, <br>**Available values:** "full-doc", "paragraph" | Parent-child segmentation mode. | No |
 | pre_processing_rules | [ [PreProcessingRule](#preprocessingrule) ] | Pre-processing rules to apply before segmentation. | No |
 | segmentation | [Segmentation](#segmentation) | Parent chunk segmentation settings. | No |
 | subchunk_segmentation | [Segmentation](#segmentation) | Child chunk segmentation settings. | No |
@@ -22477,7 +22477,7 @@ Query parameters for listing snippet published workflows.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | action | string, <br>**Available values:** "complete_task", "disable_current_workspace", "enable_current_workspace", "skip", "uncomplete_task" | State update action<br>*Enum:* `"complete_task"`, `"disable_current_workspace"`, `"enable_current_workspace"`, `"skip"`, `"uncomplete_task"` | Yes |
-| task_id | string | Task ID for task actions | No |
+| task_id | string, <br>**Available values:** "home", "integration", "knowledge", "studio" | Task ID for task actions | No |
 
 #### StepByStepTourStateResponse
 
@@ -22943,7 +22943,7 @@ Tool label
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| visibility | string | Visibility for the OAuth credential. Defaults to 'only_me'. | No |
+| visibility | string, <br>**Available values:** "all_team_members", "only_me" | Visibility for the OAuth credential. Defaults to 'only_me'. | No |
 
 #### ToolOAuthCustomClientPayload
 
@@ -23075,7 +23075,7 @@ removes TOOLS_SELECTOR from PluginParameterType
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| type | string |  | No |
+| type | string, <br>**Available values:** "api", "builtin", "mcp", "model", "workflow" |  | No |
 
 #### ToolProviderListResponse
 
@@ -23693,7 +23693,7 @@ in form definition, or a variable while the workflow is running.
 | ---- | ---- | ----------- | -------- |
 | keyword_setting | [WeightKeywordSetting](#weightkeywordsetting) | Keyword search weight settings. | No |
 | vector_setting | [WeightVectorSetting](#weightvectorsetting) | Semantic search weight settings. | No |
-| weight_type | string | Strategy for balancing semantic and keyword search weights. | No |
+| weight_type | string, <br>**Available values:** "customized", "keyword_first", "semantic_first" | Strategy for balancing semantic and keyword search weights. | No |
 
 #### WeightVectorSetting
 
@@ -24199,7 +24199,7 @@ can reuse its existing handler.
 | description | string |  | No |
 | event | string |  | No |
 | icon | string |  | No |
-| mode | string | *Enum:* `"advanced-chat"`, `"workflow"` | Yes |
+| mode | string, <br>**Available values:** "advanced-chat", "workflow" | *Enum:* `"advanced-chat"`, `"workflow"` | Yes |
 | nodes | [ [WorkflowPlanNodeResponse](#workflowplannoderesponse) ] |  | Yes |
 | start_inputs | [ [WorkflowPlanStartInputResponse](#workflowplanstartinputresponse) ] |  | No |
 | title | string |  | No |
@@ -24214,7 +24214,7 @@ can reuse its existing handler.
 | graph | [WorkflowGraph](#workflowgraph) |  | Yes |
 | icon | string |  | No |
 | message | string |  | No |
-| mode | string |  | No |
+| mode | string, <br>**Available values:** "advanced-chat", "workflow" |  | No |
 
 #### WorkflowGenerateResultEventResponse
 
@@ -24227,7 +24227,7 @@ can reuse its existing handler.
 | graph | [WorkflowGraph](#workflowgraph) |  | Yes |
 | icon | string |  | No |
 | message | string |  | No |
-| mode | string |  | No |
+| mode | string, <br>**Available values:** "advanced-chat", "workflow" |  | No |
 
 #### WorkflowGenerateStreamEventResponse
 
@@ -24527,9 +24527,9 @@ Lifecycle state for an asynchronous archive download request.
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| status | string | Workflow run status filter | No |
+| status | string, <br>**Available values:** "failed", "partial-succeeded", "running", "stopped", "succeeded" | Workflow run status filter | No |
 | time_range | string | Filter by time range (optional): e.g., 7d (7 days), 4h (4 hours), 30m (30 minutes), 30s (30 seconds). Filters by created_at field. | No |
-| triggered_from | string | Filter by trigger source: debugging or app-run. Default: debugging | No |
+| triggered_from | string, <br>**Available values:** "app-run", "debugging" | Filter by trigger source: debugging or app-run. Default: debugging | No |
 
 #### WorkflowRunCountResponse
 
@@ -24601,8 +24601,8 @@ Lifecycle state for an asynchronous archive download request.
 | ---- | ---- | ----------- | -------- |
 | last_id | string | Last run ID for pagination | No |
 | limit | integer, <br>**Default:** 20 | Number of items per page (1-100) | No |
-| status | string | Workflow run status filter | No |
-| triggered_from | string | Filter by trigger source: debugging or app-run. Default: debugging | No |
+| status | string, <br>**Available values:** "failed", "partial-succeeded", "running", "stopped", "succeeded" | Workflow run status filter | No |
+| triggered_from | string, <br>**Available values:** "app-run", "debugging" | Filter by trigger source: debugging or app-run. Default: debugging | No |
 
 #### WorkflowRunNodeExecutionListResponse
 
@@ -24900,7 +24900,7 @@ Workflow tool configuration
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| language | string | Localized policy label language | No |
+| language | string, <br>**Available values:** "en", "ja", "zh" | Localized policy label language | No |
 
 #### _AccessPolicyList
 
@@ -24959,7 +24959,7 @@ Workflow tool configuration
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| language | string | Localized policy label language | No |
+| language | string, <br>**Available values:** "en", "ja", "zh" | Localized policy label language | No |
 | limit | integer |  | No |
 | page | integer |  | No |
 | reverse | boolean |  | No |
diff --git a/api/openapi/markdown/service-openapi.md b/api/openapi/markdown/service-openapi.md
index a09bd57e255..80b5ab94db8 100644
--- a/api/openapi/markdown/service-openapi.md
+++ b/api/openapi/markdown/service-openapi.md
@@ -2587,7 +2587,7 @@ Public pause reason emitted by a blocking Chatflow execution.
 | files | [ object<br>object<br>object<br>object ] | File list for multimodal understanding, including images, documents, audio, and video. To attach a local file, first upload it via [Upload File](/api-reference/files/upload-file) and use the returned `id` as `upload_file_id` with `transfer_method: local_file`. | No |
 | inputs | object | Values for app-defined variables. Refer to the `user_input_form` field in the [Get App Parameters](/api-reference/applications/get-app-parameters) response to discover expected variable names and types. | Yes |
 | query | string | User input or question content. | Yes |
-| response_mode | string | Response mode. `streaming` uses Server-Sent Events; `blocking` returns after completion. New Agent app mode supports streaming only. When omitted, non-Agent apps run in blocking mode and new Agent apps stream. | No |
+| response_mode | string, <br>**Available values:** "blocking", "streaming" | Response mode. `streaming` uses Server-Sent Events; `blocking` returns after completion. New Agent app mode supports streaming only. When omitted, non-Agent apps run in blocking mode and new Agent apps stream. | No |
 | workflow_id | string | Published workflow version ID to execute for advanced chat. If omitted, the app's current published workflow is used. | No |
 
 #### ChatRequestPayloadWithUser
@@ -2599,7 +2599,7 @@ Public pause reason emitted by a blocking Chatflow execution.
 | files | [ object<br>object<br>object<br>object ] | File list for multimodal understanding, including images, documents, audio, and video. To attach a local file, first upload it via [Upload File](/api-reference/files/upload-file) and use the returned `id` as `upload_file_id` with `transfer_method: local_file`. | No |
 | inputs | object | Values for app-defined variables. Refer to the `user_input_form` field in the [Get App Parameters](/api-reference/applications/get-app-parameters) response to discover expected variable names and types. | Yes |
 | query | string | User input or question content. | Yes |
-| response_mode | string | Response mode. `streaming` uses Server-Sent Events; `blocking` returns after completion. New Agent app mode supports streaming only. When omitted, non-Agent apps run in blocking mode and new Agent apps stream. | No |
+| response_mode | string, <br>**Available values:** "blocking", "streaming" | Response mode. `streaming` uses Server-Sent Events; `blocking` returns after completion. New Agent app mode supports streaming only. When omitted, non-Agent apps run in blocking mode and new Agent apps stream. | No |
 | user | string | User identifier, unique within the application. This identifier scopes data access; resources created with one `user` value are only visible when queried with the same `user` value. | Yes |
 | workflow_id | string | Published workflow version ID to execute for advanced chat. If omitted, the app's current published workflow is used. | No |
 
@@ -2672,7 +2672,7 @@ Public pause reason emitted by a blocking Chatflow execution.
 | files | [ object<br>object<br>object<br>object ] | File list for multimodal understanding, including images, documents, audio, and video. To attach a local file, first upload it via [Upload File](/api-reference/files/upload-file) and use the returned `id` as `upload_file_id` with `transfer_method: local_file`. | No |
 | inputs | object | Values for app-defined variables. Refer to the `user_input_form` field in the [Get App Parameters](/api-reference/applications/get-app-parameters) response to discover expected variable names and types. | Yes |
 | query | string | User input or prompt content. | No |
-| response_mode | string | Response mode. `streaming` uses Server-Sent Events; `blocking` returns after completion. When omitted, the request runs in blocking mode. | No |
+| response_mode | string, <br>**Available values:** "blocking", "streaming" | Response mode. `streaming` uses Server-Sent Events; `blocking` returns after completion. When omitted, the request runs in blocking mode. | No |
 
 #### CompletionRequestPayloadWithUser
 
@@ -2681,7 +2681,7 @@ Public pause reason emitted by a blocking Chatflow execution.
 | files | [ object<br>object<br>object<br>object ] | File list for multimodal understanding, including images, documents, audio, and video. To attach a local file, first upload it via [Upload File](/api-reference/files/upload-file) and use the returned `id` as `upload_file_id` with `transfer_method: local_file`. | No |
 | inputs | object | Values for app-defined variables. Refer to the `user_input_form` field in the [Get App Parameters](/api-reference/applications/get-app-parameters) response to discover expected variable names and types. | Yes |
 | query | string | User input or prompt content. | No |
-| response_mode | string | Response mode. `streaming` uses Server-Sent Events; `blocking` returns after completion. When omitted, the request runs in blocking mode. | No |
+| response_mode | string, <br>**Available values:** "blocking", "streaming" | Response mode. `streaming` uses Server-Sent Events; `blocking` returns after completion. When omitted, the request runs in blocking mode. | No |
 | user | string | User identifier, unique within the application. This identifier scopes data access; resources created with one `user` value are only visible when queried with the same `user` value. | Yes |
 
 #### Condition
@@ -2797,7 +2797,7 @@ Enum class for custom configuration status.
 | embedding_model_provider | string | Embedding model provider. Use the `provider` field from [Get Available Models](/api-reference/models/get-available-models) with `model_type=text-embedding`. | No |
 | external_knowledge_api_id | string | ID of the external knowledge API. | No |
 | external_knowledge_id | string | ID of the external knowledge base. | No |
-| indexing_technique | string | `high_quality` uses embedding models for precise search; `economy` uses keyword-based indexing. | No |
+| indexing_technique | string, <br>**Available values:** "economy", "high_quality" | `high_quality` uses embedding models for precise search; `economy` uses keyword-based indexing. | No |
 | name | string | Name of the knowledge base. | Yes |
 | permission | [PermissionEnum](#permissionenum) | Controls who can access this knowledge base. `only_me` restricts access to the creator, `all_team_members` grants workspace-wide access, and `partial_members` grants access to specified members. | No |
 | provider | string, <br>**Available values:** "external", "vendor", <br>**Default:** vendor | Knowledge base provider: `vendor` for internal knowledge bases, `external` for external ones.<br>*Enum:* `"external"`, `"vendor"` | No |
@@ -3039,7 +3039,7 @@ Enum class for custom configuration status.
 | external_knowledge_api_id | string | ID of the external knowledge API. | No |
 | external_knowledge_id | string | ID of the external knowledge base. | No |
 | external_retrieval_model | object | Retrieval settings for external knowledge bases. | No |
-| indexing_technique | string | `high_quality` uses embedding models for precise search; `economy` uses keyword-based indexing. | No |
+| indexing_technique | string, <br>**Available values:** "economy", "high_quality" | `high_quality` uses embedding models for precise search; `economy` uses keyword-based indexing. | No |
 | name | string | Name of the knowledge base. | No |
 | partial_member_list | [ object ] | List of team members with access when `permission` is `partial_members`. | No |
 | permission | [PermissionEnum](#permissionenum) | Controls who can access this knowledge base. `only_me` restricts access to the creator, `all_team_members` grants workspace-wide access, and `partial_members` grants access to specified members. | No |
@@ -3167,7 +3167,7 @@ Request payload for bulk downloading documents as a zip archive.
 | keyword | string | Search keyword to filter by document name. | No |
 | limit | integer, <br>**Default:** 20 | Number of items per page. Server caps at `100`. | No |
 | page | integer, <br>**Default:** 1 | Page number to retrieve. | No |
-| status | string | Filter by display status. | No |
+| status | string, <br>**Available values:** "archived", "available", "disabled", "error", "indexing", "paused", "queuing" | Filter by display status. | No |
 
 #### DocumentListResponse
 
@@ -3265,7 +3265,7 @@ Request payload for bulk downloading documents as a zip archive.
 | doc_language | string, <br>**Default:** English | Language of the document for processing optimization. | No |
 | embedding_model | string | Embedding model name. Use the `model` field from [Get Available Models](/api-reference/models/get-available-models) with `model_type=text-embedding`. | No |
 | embedding_model_provider | string | Embedding model provider. Use the `provider` field from [Get Available Models](/api-reference/models/get-available-models) with `model_type=text-embedding`. | No |
-| indexing_technique | string | `high_quality` uses embedding models for precise search; `economy` uses keyword-based indexing. Required when adding the first document to a knowledge base; subsequent documents inherit the knowledge base's indexing technique if omitted. | No |
+| indexing_technique | string, <br>**Available values:** "economy", "high_quality" | `high_quality` uses embedding models for precise search; `economy` uses keyword-based indexing. Required when adding the first document to a knowledge base; subsequent documents inherit the knowledge base's indexing technique if omitted. | No |
 | name | string | Document name. | Yes |
 | original_document_id | string | Original document ID for replacement. | No |
 | process_rule | [ProcessRule](#processrule) | Processing rules for chunking. | No |
@@ -3614,14 +3614,14 @@ Model class for i18n object.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | content | string | Optional text feedback providing additional detail. | No |
-| rating | string | Feedback rating. Set to `null` to revoke previously submitted feedback. | No |
+| rating | string, <br>**Available values:** "dislike", "like" | Feedback rating. Set to `null` to revoke previously submitted feedback. | No |
 
 #### MessageFeedbackPayloadWithUser
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | content | string | Optional text feedback providing additional detail. | No |
-| rating | string | Feedback rating. Set to `null` to revoke previously submitted feedback. | No |
+| rating | string, <br>**Available values:** "dislike", "like" | Feedback rating. Set to `null` to revoke previously submitted feedback. | No |
 | user | string | User identifier, unique within the application. This identifier scopes data access; resources created with one `user` value are only visible when queried with the same `user` value. | Yes |
 
 #### MessageFile
@@ -3701,7 +3701,7 @@ Metadata Filtering Condition.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | conditions | [ [Condition](#condition) ] | List of metadata conditions to evaluate. | No |
-| logical_operator | string | How to combine multiple conditions. | No |
+| logical_operator | string, <br>**Available values:** "and", "or" | How to combine multiple conditions. | No |
 
 #### MetadataOperationData
 
@@ -3935,7 +3935,7 @@ Model class for provider with models response.
 | ---- | ---- | ----------- | -------- |
 | metadata_filtering_conditions | [MetadataFilteringCondition](#metadatafilteringcondition) | Restrict retrieval to chunks whose document metadata matches the given conditions. Conditions are evaluated server-side against document metadata fields. | No |
 | reranking_enable | boolean | Whether reranking is enabled. | Yes |
-| reranking_mode | string | Reranking mode. Required when `reranking_enable` is `true`. | No |
+| reranking_mode | string, <br>**Available values:** "reranking_model", "weighted_score" | Reranking mode. Required when `reranking_enable` is `true`. | No |
 | reranking_model | [RerankingModel](#rerankingmodel) | Reranking model configuration. | No |
 | score_threshold | number | Minimum similarity score for results. Only effective when score threshold filtering is enabled. | No |
 | score_threshold_enabled | boolean | Whether score threshold filtering is enabled. | Yes |
@@ -3969,7 +3969,7 @@ Model class for provider with models response.
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| parent_mode | string | Parent-child segmentation mode. | No |
+| parent_mode | string, <br>**Available values:** "full-doc", "paragraph" | Parent-child segmentation mode. | No |
 | pre_processing_rules | [ [PreProcessingRule](#preprocessingrule) ] | Pre-processing rules to apply before segmentation. | No |
 | segmentation | [Segmentation](#segmentation) | Parent chunk segmentation settings. | No |
 | subchunk_segmentation | [Segmentation](#segmentation) | Child chunk segmentation settings. | No |
@@ -4300,7 +4300,7 @@ in form definition, or a variable while the workflow is running.
 | ---- | ---- | ----------- | -------- |
 | keyword_setting | [WeightKeywordSetting](#weightkeywordsetting) | Keyword search weight settings. | No |
 | vector_setting | [WeightVectorSetting](#weightvectorsetting) | Semantic search weight settings. | No |
-| weight_type | string | Strategy for balancing semantic and keyword search weights. | No |
+| weight_type | string, <br>**Available values:** "customized", "keyword_first", "semantic_first" | Strategy for balancing semantic and keyword search weights. | No |
 
 #### WeightVectorSetting
 
@@ -4383,7 +4383,7 @@ Blocking workflow response for a finished or paused execution.
 | keyword | string | Keyword to search in logs. | No |
 | limit | integer, <br>**Default:** 20 | Number of items per page. | No |
 | page | integer, <br>**Default:** 1 | Page number for pagination. | No |
-| status | string | Filter by execution status. | No |
+| status | string, <br>**Available values:** "failed", "stopped", "succeeded" | Filter by execution status. | No |
 
 #### WorkflowPauseReasonResponse
 
@@ -4452,7 +4452,7 @@ Public pause reason emitted by a blocking Workflow execution.
 | ---- | ---- | ----------- | -------- |
 | files | [ object<br>object<br>object<br>object ] | File list for workflow system file inputs. Available when file upload is enabled for the workflow. To attach a local file, first upload it via [Upload File](/api-reference/files/upload-file) and use the returned `id` as `upload_file_id` with `transfer_method: local_file`. | No |
 | inputs | object | Key-value pairs for workflow input variables. Values for file-type variables should be arrays of file objects with `type`, `transfer_method`, and either `url` or `upload_file_id`. Refer to the `user_input_form` field in the [Get App Parameters](/api-reference/applications/get-app-parameters) response to discover the variable names and types expected by your app. | Yes |
-| response_mode | string | Response mode. Use `blocking` for synchronous responses or `streaming` for Server-Sent Events. When omitted, the request runs in blocking mode. | No |
+| response_mode | string, <br>**Available values:** "blocking", "streaming" | Response mode. Use `blocking` for synchronous responses or `streaming` for Server-Sent Events. When omitted, the request runs in blocking mode. | No |
 
 #### WorkflowRunPayloadWithUser
 
@@ -4460,7 +4460,7 @@ Public pause reason emitted by a blocking Workflow execution.
 | ---- | ---- | ----------- | -------- |
 | files | [ object<br>object<br>object<br>object ] | File list for workflow system file inputs. Available when file upload is enabled for the workflow. To attach a local file, first upload it via [Upload File](/api-reference/files/upload-file) and use the returned `id` as `upload_file_id` with `transfer_method: local_file`. | No |
 | inputs | object | Key-value pairs for workflow input variables. Values for file-type variables should be arrays of file objects with `type`, `transfer_method`, and either `url` or `upload_file_id`. Refer to the `user_input_form` field in the [Get App Parameters](/api-reference/applications/get-app-parameters) response to discover the variable names and types expected by your app. | Yes |
-| response_mode | string | Response mode. Use `blocking` for synchronous responses or `streaming` for Server-Sent Events. When omitted, the request runs in blocking mode. | No |
+| response_mode | string, <br>**Available values:** "blocking", "streaming" | Response mode. Use `blocking` for synchronous responses or `streaming` for Server-Sent Events. When omitted, the request runs in blocking mode. | No |
 | user | string | User identifier, unique within the application. This identifier scopes data access; resources created with one `user` value are only visible when queried with the same `user` value. | Yes |
 
 #### WorkflowRunResponse
diff --git a/api/openapi/markdown/web-openapi.md b/api/openapi/markdown/web-openapi.md
index e534fe39350..3cc99e099ce 100644
--- a/api/openapi/markdown/web-openapi.md
+++ b/api/openapi/markdown/web-openapi.md
@@ -1019,7 +1019,7 @@ Button styles for user actions.
 | inputs | object | Input variables for the chat | Yes |
 | parent_message_id | string | Parent message ID | No |
 | query | string | User query/message | Yes |
-| response_mode | string | Response mode: blocking or streaming | No |
+| response_mode | string, <br>**Available values:** "blocking", "streaming" | Response mode: blocking or streaming | No |
 | retriever_from | string, <br>**Default:** web_app | Source of retriever | No |
 
 #### CompletionMessagePayload
@@ -1029,7 +1029,7 @@ Button styles for user actions.
 | files | [ object ] | Files to be processed | No |
 | inputs | object | Input variables for the completion | Yes |
 | query | string | Query text for completion | No |
-| response_mode | string | Response mode: blocking or streaming | No |
+| response_mode | string, <br>**Available values:** "blocking", "streaming" | Response mode: blocking or streaming | No |
 | retriever_from | string, <br>**Default:** web_app | Source of retriever | No |
 
 #### ConversationInfiniteScrollPagination
@@ -1322,7 +1322,7 @@ Parsed multipart form fields for HITL uploads.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | content | string | Optional text feedback providing additional detail. | No |
-| rating | string | Feedback rating. Set to `null` to revoke previously submitted feedback. | No |
+| rating | string, <br>**Available values:** "dislike", "like" | Feedback rating. Set to `null` to revoke previously submitted feedback. | No |
 
 #### MessageFile
 
diff --git a/api/repositories/step_by_step_tour_repository.py b/api/repositories/step_by_step_tour_repository.py
new file mode 100644
index 00000000000..7dc7a6d6bf2
--- /dev/null
+++ b/api/repositories/step_by_step_tour_repository.py
@@ -0,0 +1,189 @@
+"""SQLAlchemy repository for account Step-by-step Tour state."""
+
+import logging
+from collections.abc import Callable
+from typing import Protocol, override, runtime_checkable
+
+from sqlalchemy import select, update
+from sqlalchemy.exc import IntegrityError, OperationalError
+from sqlalchemy.orm import Session, sessionmaker
+
+from models.onboarding import AccountStepByStepTourState
+from services.entities.onboarding_entities import StepByStepTourState
+from services.step_by_step_tour_service import StepByStepTourStateRepository
+
+logger = logging.getLogger(__name__)
+
+_MYSQL_RETRYABLE_LOCK_ERRNOS = frozenset({1205, 1213})
+_MAX_LOCK_ATTEMPTS = 3
+
+
+@runtime_checkable
+class _ErrorWithErrno(Protocol):
+    @property
+    def errno(self) -> object: ...
+
+
+class SQLAlchemyStepByStepTourStateRepository(StepByStepTourStateRepository):
+    def __init__(self, session_factory: sessionmaker[Session]) -> None:
+        self._session_factory = session_factory
+
+    @override
+    def get(self, account_id: str) -> StepByStepTourState | None:
+        with self._session_factory() as session:
+            model = self._get_model(account_id, session=session)
+            return self._to_state(model) if model is not None else None
+
+    @override
+    def initialize(self, account_id: str, first_workspace_id: str) -> StepByStepTourState:
+        """Create state with its first workspace, or atomically claim a legacy empty state."""
+        return self._run_with_lock_retry(
+            lambda: self._initialize_once(account_id, first_workspace_id),
+        )
+
+    def _initialize_once(self, account_id: str, first_workspace_id: str) -> StepByStepTourState:
+        with self._session_factory() as session:
+            model = self._get_model(account_id, session=session)
+            if model is None:
+                model = AccountStepByStepTourState(
+                    account_id=account_id,
+                    first_workspace_id=first_workspace_id,
+                )
+                session.add(model)
+                try:
+                    session.commit()
+                except IntegrityError:
+                    # A concurrent request inserted the account-owned row first.
+                    session.rollback()
+                    model = self._get_model(account_id, session=session)
+                    if model is None:
+                        raise
+                else:
+                    session.refresh(model)
+                    return self._to_state(model)
+
+            if model.first_workspace_id is None:
+                stmt = (
+                    update(AccountStepByStepTourState)
+                    .where(
+                        AccountStepByStepTourState.account_id == account_id,
+                        AccountStepByStepTourState.first_workspace_id.is_(None),
+                    )
+                    .values(first_workspace_id=first_workspace_id)
+                    .execution_options(synchronize_session=False)
+                )
+                session.execute(stmt)
+                session.commit()
+                # A competing conditional update may have won while this request waited.
+                session.refresh(model)
+
+            return self._to_state(model)
+
+    @override
+    def mutate(
+        self,
+        account_id: str,
+        mutation: Callable[[StepByStepTourState], StepByStepTourState],
+    ) -> StepByStepTourState:
+        """Lock, create if needed, mutate, and persist account state in one transaction."""
+        return self._run_with_lock_retry(
+            lambda: self._mutate_once(account_id, mutation),
+        )
+
+    def _mutate_once(
+        self,
+        account_id: str,
+        mutation: Callable[[StepByStepTourState], StepByStepTourState],
+    ) -> StepByStepTourState:
+        with self._session_factory() as session:
+            # Probe without a locking read so a missing MySQL unique key does not
+            # acquire a gap/next-key lock before the insert.
+            model = self._get_model(account_id, session=session)
+            if model is None:
+                model = AccountStepByStepTourState(account_id=account_id)
+                session.add(model)
+                try:
+                    session.flush()
+                except IntegrityError:
+                    # A concurrent mutation created the row. Start a new transaction,
+                    # lock its committed state, and replay the pure mutation on it.
+                    session.rollback()
+                    model = self._get_model(account_id, session=session, lock_for_update=True)
+                    if model is None:
+                        raise
+            else:
+                model = self._get_model(account_id, session=session, lock_for_update=True)
+                if model is None:
+                    raise RuntimeError("Step-by-step Tour state disappeared while acquiring its lock")
+
+            state = mutation(self._to_state(model))
+            if state.account_id != account_id:
+                raise ValueError("Step-by-step Tour mutation cannot change account ownership")
+            # first_workspace_id is write-once and owned exclusively by initialize().
+            model.skipped = state.skipped
+            model.completed_task_ids = list(state.completed_task_ids)
+            model.manually_enabled_workspace_ids = list(state.manually_enabled_workspace_ids)
+            model.manually_disabled_workspace_ids = list(state.manually_disabled_workspace_ids)
+            session.commit()
+            session.refresh(model)
+            return self._to_state(model)
+
+    @staticmethod
+    def _run_with_lock_retry[T](operation: Callable[[], T]) -> T:
+        for attempt in range(1, _MAX_LOCK_ATTEMPTS):
+            try:
+                return operation()
+            except OperationalError as exc:
+                if not _is_retryable_mysql_lock_error(exc):
+                    raise
+                logger.warning(
+                    "Retrying Step-by-step Tour transaction after MySQL lock failure (attempt %s/%s)",
+                    attempt,
+                    _MAX_LOCK_ATTEMPTS,
+                )
+        return operation()
+
+    @staticmethod
+    def _get_model(
+        account_id: str,
+        *,
+        session: Session,
+        lock_for_update: bool = False,
+    ) -> AccountStepByStepTourState | None:
+        stmt = select(AccountStepByStepTourState).where(AccountStepByStepTourState.account_id == account_id).limit(1)
+        if lock_for_update:
+            stmt = stmt.with_for_update().execution_options(populate_existing=True)
+        return session.execute(stmt).scalar_one_or_none()
+
+    @staticmethod
+    def _to_state(model: AccountStepByStepTourState) -> StepByStepTourState:
+        return StepByStepTourState(
+            account_id=model.account_id,
+            first_workspace_id=model.first_workspace_id,
+            skipped=model.skipped,
+            completed_task_ids=tuple(model.completed_task_ids),
+            manually_enabled_workspace_ids=tuple(model.manually_enabled_workspace_ids),
+            manually_disabled_workspace_ids=tuple(model.manually_disabled_workspace_ids),
+            updated_at=model.updated_at,
+        )
+
+
+def _is_retryable_mysql_lock_error(exc: OperationalError) -> bool:
+    orig = exc.orig
+    if isinstance(orig, _ErrorWithErrno) and _is_retryable_mysql_lock_error_code(orig.errno):
+        return True
+    if not isinstance(orig, BaseException) or not orig.args:
+        return False
+    return _is_retryable_mysql_lock_error_code(orig.args[0])
+
+
+def _is_retryable_mysql_lock_error_code(candidate: object) -> bool:
+    if isinstance(candidate, bool):
+        return False
+    if isinstance(candidate, int):
+        code = candidate
+    elif isinstance(candidate, str) and candidate.isdecimal():
+        code = int(candidate)
+    else:
+        return False
+    return code in _MYSQL_RETRYABLE_LOCK_ERRNOS
diff --git a/api/services/entities/notification_entities.py b/api/services/entities/notification_entities.py
new file mode 100644
index 00000000000..6686c5edb99
--- /dev/null
+++ b/api/services/entities/notification_entities.py
@@ -0,0 +1,38 @@
+"""Framework-independent notification contracts."""
+
+from collections.abc import Mapping
+from typing import NamedTuple
+
+
+class NotificationContent(NamedTuple):
+    lang: str
+    title: str
+    subtitle: str
+    body: str
+    title_pic_url: str
+
+
+class AccountNotification(NamedTuple):
+    notification_id: str | None
+    frequency: str | None
+    contents: Mapping[str, NotificationContent]
+
+
+class AccountNotificationBatch(NamedTuple):
+    should_show: bool
+    notifications: tuple[AccountNotification, ...]
+
+
+class NotificationItem(NamedTuple):
+    notification_id: str | None
+    frequency: str | None
+    lang: str
+    title: str
+    subtitle: str
+    body: str
+    title_pic_url: str
+
+
+class NotificationResult(NamedTuple):
+    should_show: bool
+    notifications: tuple[NotificationItem, ...]
diff --git a/api/services/entities/onboarding_entities.py b/api/services/entities/onboarding_entities.py
new file mode 100644
index 00000000000..2550db489e4
--- /dev/null
+++ b/api/services/entities/onboarding_entities.py
@@ -0,0 +1,42 @@
+"""Framework-independent Step-by-step Tour contracts."""
+
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Literal, TypeAlias
+
+# Assignment-form aliases preserve Literal enum values in Pydantic-generated OpenAPI schemas.
+StepByStepTourAction: TypeAlias = Literal[  # noqa: UP040
+    "skip",
+    "complete_task",
+    "uncomplete_task",
+    "enable_current_workspace",
+    "disable_current_workspace",
+]
+StepByStepTourTaskId: TypeAlias = Literal["home", "studio", "knowledge", "integration"]  # noqa: UP040
+
+
+@dataclass(frozen=True, slots=True)
+class StepByStepTourPatch:
+    action: StepByStepTourAction
+    task_id: StepByStepTourTaskId | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class StepByStepTourState:
+    account_id: str
+    first_workspace_id: str | None = None
+    skipped: bool = False
+    completed_task_ids: tuple[str, ...] = ()
+    manually_enabled_workspace_ids: tuple[str, ...] = ()
+    manually_disabled_workspace_ids: tuple[str, ...] = ()
+    updated_at: datetime | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class StepByStepTourResult:
+    first_workspace_id: str | None = None
+    skipped: bool = False
+    completed_task_ids: tuple[str, ...] = ()
+    manually_enabled_workspace_ids: tuple[str, ...] = ()
+    manually_disabled_workspace_ids: tuple[str, ...] = ()
+    updated_at: datetime | None = None
diff --git a/api/services/notification_gateway.py b/api/services/notification_gateway.py
new file mode 100644
index 00000000000..cb7cc5e74d0
--- /dev/null
+++ b/api/services/notification_gateway.py
@@ -0,0 +1,48 @@
+"""Billing-backed notification gateway."""
+
+from collections.abc import Mapping
+from typing import Any, override
+
+from services.billing_service import BillingService
+from services.entities.notification_entities import (
+    AccountNotification,
+    AccountNotificationBatch,
+    NotificationContent,
+)
+from services.notification_service import NotificationGateway
+
+
+class BillingNotificationGateway(NotificationGateway):
+    @override
+    def get_active(self, account_id: str) -> AccountNotificationBatch:
+        payload = BillingService.get_account_notification(account_id)
+        notifications = tuple(self._map_notification(item) for item in payload.get("notifications") or ())
+        return AccountNotificationBatch(
+            should_show=bool(payload.get("shouldShow")),
+            notifications=notifications,
+        )
+
+    @override
+    def dismiss(self, notification_id: str, account_id: str) -> None:
+        BillingService.dismiss_notification(notification_id=notification_id, account_id=account_id)
+
+    @classmethod
+    def _map_notification(cls, payload: Mapping[str, Any]) -> AccountNotification:
+        raw_contents = payload.get("contents") or {}
+        contents = {language: cls._map_content(content) for language, content in raw_contents.items() if content}
+        return AccountNotification(
+            notification_id=payload.get("notificationId"),
+            frequency=payload.get("frequency"),
+            contents=contents,
+        )
+
+    @staticmethod
+    def _map_content(payload: Mapping[str, Any]) -> NotificationContent:
+        return NotificationContent(
+            # The application service owns the requested-language fallback.
+            lang=payload.get("lang") or "",
+            title=payload.get("title") or "",
+            subtitle=payload.get("subtitle") or "",
+            body=payload.get("body") or "",
+            title_pic_url=payload.get("titlePicUrl") or "",
+        )
diff --git a/api/services/notification_service.py b/api/services/notification_service.py
new file mode 100644
index 00000000000..13236ef16ed
--- /dev/null
+++ b/api/services/notification_service.py
@@ -0,0 +1,60 @@
+"""Application service for Console account notifications."""
+
+from typing import Protocol
+
+from machinery.context import RequestContext
+from services.account_ports import AccountRepository
+from services.entities.notification_entities import (
+    AccountNotification,
+    AccountNotificationBatch,
+    NotificationContent,
+    NotificationItem,
+    NotificationResult,
+)
+
+_FALLBACK_LANGUAGE = "en-US"
+
+
+class NotificationGateway(Protocol):
+    def get_active(self, account_id: str) -> AccountNotificationBatch: ...
+
+    def dismiss(self, notification_id: str, account_id: str) -> None: ...
+
+
+class NotificationService:
+    def __init__(self, *, accounts: AccountRepository, notifications: NotificationGateway) -> None:
+        self._accounts = accounts
+        self._notifications = notifications
+
+    def get_active(self, context: RequestContext) -> NotificationResult:
+        batch = self._notifications.get_active(context.account_id)
+        if not batch.should_show:
+            return NotificationResult(should_show=False, notifications=())
+
+        account = self._accounts.get(context.account_id)
+        if account is None:
+            raise RuntimeError("Console account admission resolved an unknown account")
+        language = account.interface_language or _FALLBACK_LANGUAGE
+
+        notifications = tuple(self._localize(notification, language) for notification in batch.notifications)
+        return NotificationResult(should_show=bool(notifications), notifications=notifications)
+
+    def dismiss(self, context: RequestContext, notification_id: str) -> None:
+        self._notifications.dismiss(notification_id, context.account_id)
+
+    @staticmethod
+    def _localize(notification: AccountNotification, language: str) -> NotificationItem:
+        content = (
+            notification.contents.get(language)
+            or notification.contents.get(_FALLBACK_LANGUAGE)
+            or next(iter(notification.contents.values()), NotificationContent(language, "", "", "", ""))
+        )
+        return NotificationItem(
+            notification_id=notification.notification_id,
+            frequency=notification.frequency,
+            lang=content.lang or language,
+            title=content.title,
+            subtitle=content.subtitle,
+            body=content.body,
+            title_pic_url=content.title_pic_url,
+        )
diff --git a/api/services/step_by_step_tour_service.py b/api/services/step_by_step_tour_service.py
index b01d59c1acc..9597d3d5e77 100644
--- a/api/services/step_by_step_tour_service.py
+++ b/api/services/step_by_step_tour_service.py
@@ -1,221 +1,161 @@
-"""Account-level Step-by-step Tour persistence."""
+"""Application service for account-level Step-by-step Tour use cases."""
 
+from collections.abc import Callable
+from dataclasses import replace
 from datetime import datetime
-from typing import NotRequired, TypedDict
+from typing import Protocol, get_args
 
-from sqlalchemy import select
-from sqlalchemy.exc import IntegrityError
-from sqlalchemy.orm import Session, scoped_session
-
-from configs import dify_config
 from libs.datetime_utils import ensure_naive_utc
-from models.account import Account
-from models.onboarding import AccountStepByStepTourState
+from machinery.context import RequestContext
+from services.account_ports import AccountRepository
+from services.entities.onboarding_entities import (
+    StepByStepTourPatch,
+    StepByStepTourResult,
+    StepByStepTourState,
+    StepByStepTourTaskId,
+)
 
-STEP_BY_STEP_TOUR_TASK_IDS = frozenset(("home", "studio", "knowledge", "integration"))
+_TASK_IDS: frozenset[str] = frozenset(get_args(StepByStepTourTaskId))
 
 
-class StepByStepTourStateResponse(TypedDict):
-    first_workspace_id: str | None
-    skipped: bool
-    completed_task_ids: list[str]
-    manually_enabled_workspace_ids: list[str]
-    manually_disabled_workspace_ids: list[str]
-    updated_at: datetime | None
+class StepByStepTourStateRepository(Protocol):
+    def get(self, account_id: str) -> StepByStepTourState | None: ...
 
+    def initialize(self, account_id: str, first_workspace_id: str) -> StepByStepTourState: ...
 
-class StepByStepTourPatch(TypedDict):
-    action: str
-    task_id: NotRequired[str | None]
+    def mutate(
+        self,
+        account_id: str,
+        mutation: Callable[[StepByStepTourState], StepByStepTourState],
+    ) -> StepByStepTourState: ...
 
 
 class StepByStepTourService:
-    """Coordinate persisted tour state with account eligibility rules."""
-
-    @classmethod
-    def get_state(
-        cls,
+    def __init__(
+        self,
         *,
-        account: Account,
-        current_tenant_id: str,
-        session: Session | scoped_session,
-    ) -> StepByStepTourStateResponse:
-        eligible = cls.is_eligible(account)
-        state = cls._get_state(account.id, session=session)
+        accounts: AccountRepository,
+        states: StepByStepTourStateRepository,
+        enabled: bool,
+        rollout_started_at: datetime | None,
+    ) -> None:
+        self._accounts = accounts
+        self._states = states
+        self._enabled = enabled
+        self._rollout_started_at = rollout_started_at
 
-        if eligible:
-            state = cls._ensure_state(account.id, session=session, state=state)
-            if state.first_workspace_id is None:
-                state.first_workspace_id = current_tenant_id
-                session.commit()
-                session.refresh(state)
+    def get_state(self, context: RequestContext) -> StepByStepTourResult:
+        workspace_id = self._require_workspace(context)
+        account = self._accounts.get(context.account_id)
+        if account is None:
+            raise RuntimeError("Console account admission resolved an unknown account")
 
-        return cls._build_response(state=state)
+        if not self._is_eligible(account.initialized_at or account.created_at):
+            return self._to_result(self._states.get(context.account_id))
 
-    @classmethod
-    def patch_state(
-        cls,
-        *,
-        account: Account,
-        current_tenant_id: str,
-        patch: StepByStepTourPatch,
-        session: Session | scoped_session,
-    ) -> StepByStepTourStateResponse:
-        state = cls._ensure_state(account.id, session=session, state=None)
-        cls._apply_action(
-            state=state,
-            action=patch["action"],
-            task_id=patch.get("task_id"),
-            current_tenant_id=current_tenant_id,
+        return self._to_result(self._states.initialize(context.account_id, workspace_id))
+
+    def patch_state(self, context: RequestContext, patch: StepByStepTourPatch) -> StepByStepTourResult:
+        workspace_id = self._require_workspace(context)
+        state = self._states.mutate(
+            context.account_id,
+            lambda current: self._apply_action(current, patch=patch, workspace_id=workspace_id),
         )
+        return self._to_result(state)
 
-        session.commit()
-        session.refresh(state)
-        return cls._build_response(state=state)
-
-    @classmethod
-    def is_eligible(cls, account: Account) -> bool:
-        if not dify_config.ENABLE_STEP_BY_STEP_TOUR:
+    def _is_eligible(self, account_started_at: datetime) -> bool:
+        if not self._enabled or self._rollout_started_at is None:
             return False
-
-        rollout_started_at = dify_config.STEP_BY_STEP_TOUR_ROLLOUT_STARTED_AT
-        if rollout_started_at is None:
-            return False
-
-        account_started_at = account.initialized_at or account.created_at
-        if account_started_at is None:
-            return False
-
-        return ensure_naive_utc(account_started_at) >= ensure_naive_utc(rollout_started_at)
-
-    @classmethod
-    def _get_state(
-        cls,
-        account_id: str,
-        *,
-        session: Session | scoped_session,
-    ) -> AccountStepByStepTourState | None:
-        stmt = select(AccountStepByStepTourState).where(AccountStepByStepTourState.account_id == account_id).limit(1)
-        return session.execute(stmt).scalar_one_or_none()
-
-    @classmethod
-    def _ensure_state(
-        cls,
-        account_id: str,
-        *,
-        session: Session | scoped_session,
-        state: AccountStepByStepTourState | None,
-    ) -> AccountStepByStepTourState:
-        if state is None:
-            state = cls._get_state(account_id, session=session)
-        if state is not None:
-            return state
-
-        state = AccountStepByStepTourState(account_id=account_id)
-        session.add(state)
-        try:
-            session.flush()
-        except IntegrityError:
-            # Another tab/device can create the account row between our read and insert.
-            session.rollback()
-            state = cls._get_state(account_id, session=session)
-            if state is None:
-                raise
-        return state
+        return ensure_naive_utc(account_started_at) >= ensure_naive_utc(self._rollout_started_at)
 
     @classmethod
     def _apply_action(
         cls,
+        state: StepByStepTourState,
         *,
-        state: AccountStepByStepTourState,
-        action: str,
-        task_id: str | None,
-        current_tenant_id: str,
-    ) -> None:
-        match action:
+        patch: StepByStepTourPatch,
+        workspace_id: str,
+    ) -> StepByStepTourState:
+        match patch.action:
             case "skip":
-                state.skipped = True
-                state.manually_enabled_workspace_ids = cls._remove_id(
-                    state.manually_enabled_workspace_ids,
-                    current_tenant_id,
+                return replace(
+                    state,
+                    skipped=True,
+                    manually_enabled_workspace_ids=cls._remove_id(
+                        state.manually_enabled_workspace_ids,
+                        workspace_id,
+                    ),
                 )
             case "complete_task":
-                if task_id is None:
-                    raise ValueError("task_id is required")
-                cls._validate_task_id(task_id)
-                state.completed_task_ids = cls._add_id(state.completed_task_ids, task_id)
+                task_id = cls._require_task_id(patch.task_id)
+                return replace(state, completed_task_ids=cls._add_id(state.completed_task_ids, task_id))
             case "uncomplete_task":
-                if task_id is None:
-                    raise ValueError("task_id is required")
-                cls._validate_task_id(task_id)
-                state.completed_task_ids = cls._remove_id(state.completed_task_ids, task_id)
+                task_id = cls._require_task_id(patch.task_id)
+                return replace(state, completed_task_ids=cls._remove_id(state.completed_task_ids, task_id))
             case "enable_current_workspace":
-                state.skipped = False
-                state.manually_enabled_workspace_ids = cls._add_id(
-                    state.manually_enabled_workspace_ids,
-                    current_tenant_id,
-                )
-                state.manually_disabled_workspace_ids = cls._remove_id(
-                    state.manually_disabled_workspace_ids,
-                    current_tenant_id,
+                return replace(
+                    state,
+                    skipped=False,
+                    manually_enabled_workspace_ids=cls._add_id(
+                        state.manually_enabled_workspace_ids,
+                        workspace_id,
+                    ),
+                    manually_disabled_workspace_ids=cls._remove_id(
+                        state.manually_disabled_workspace_ids,
+                        workspace_id,
+                    ),
                 )
             case "disable_current_workspace":
-                state.manually_enabled_workspace_ids = cls._remove_id(
-                    state.manually_enabled_workspace_ids,
-                    current_tenant_id,
-                )
-                state.manually_disabled_workspace_ids = cls._add_id(
-                    state.manually_disabled_workspace_ids,
-                    current_tenant_id,
+                return replace(
+                    state,
+                    manually_enabled_workspace_ids=cls._remove_id(
+                        state.manually_enabled_workspace_ids,
+                        workspace_id,
+                    ),
+                    manually_disabled_workspace_ids=cls._add_id(
+                        state.manually_disabled_workspace_ids,
+                        workspace_id,
+                    ),
                 )
             case _:
-                raise ValueError(f"Unsupported action: {action}")
-
-    @classmethod
-    def _build_response(
-        cls,
-        *,
-        state: AccountStepByStepTourState | None,
-    ) -> StepByStepTourStateResponse:
-        if state is None:
-            return {
-                "first_workspace_id": None,
-                "skipped": False,
-                "completed_task_ids": [],
-                "manually_enabled_workspace_ids": [],
-                "manually_disabled_workspace_ids": [],
-                "updated_at": None,
-            }
-
-        return {
-            "first_workspace_id": state.first_workspace_id,
-            "skipped": state.skipped,
-            "completed_task_ids": cls._normalize_ids(state.completed_task_ids),
-            "manually_enabled_workspace_ids": cls._normalize_ids(state.manually_enabled_workspace_ids),
-            "manually_disabled_workspace_ids": cls._normalize_ids(state.manually_disabled_workspace_ids),
-            "updated_at": state.updated_at,
-        }
+                raise ValueError(f"Unsupported action: {patch.action}")
 
     @staticmethod
-    def _validate_task_id(task_id: str) -> None:
-        if task_id not in STEP_BY_STEP_TOUR_TASK_IDS:
+    def _require_workspace(context: RequestContext) -> str:
+        if context.active_workspace_id is None:
+            raise RuntimeError("Console account admission did not resolve an active workspace")
+        return context.active_workspace_id
+
+    @staticmethod
+    def _require_task_id(task_id: str | None) -> str:
+        if task_id is None:
+            raise ValueError("task_id is required")
+        if task_id not in _TASK_IDS:
             raise ValueError(f"Unsupported task_id: {task_id}")
+        return task_id
 
     @classmethod
-    def _add_id(cls, values: list[str], value: str) -> list[str]:
+    def _add_id(cls, values: tuple[str, ...], value: str) -> tuple[str, ...]:
         normalized = cls._normalize_ids(values)
-        if value in normalized:
-            return normalized
-        return [*normalized, value]
+        return normalized if value in normalized else (*normalized, value)
 
     @classmethod
-    def _remove_id(cls, values: list[str], value: str) -> list[str]:
-        return [item for item in cls._normalize_ids(values) if item != value]
+    def _remove_id(cls, values: tuple[str, ...], value: str) -> tuple[str, ...]:
+        return tuple(item for item in cls._normalize_ids(values) if item != value)
 
     @staticmethod
-    def _normalize_ids(values: list[str]) -> list[str]:
-        normalized: list[str] = []
-        for value in values:
-            if value not in normalized:
-                normalized.append(value)
-        return normalized
+    def _normalize_ids(values: tuple[str, ...]) -> tuple[str, ...]:
+        return tuple(dict.fromkeys(values))
+
+    @staticmethod
+    def _to_result(state: StepByStepTourState | None) -> StepByStepTourResult:
+        if state is None:
+            return StepByStepTourResult()
+        return StepByStepTourResult(
+            first_workspace_id=state.first_workspace_id,
+            skipped=state.skipped,
+            completed_task_ids=tuple(dict.fromkeys(state.completed_task_ids)),
+            manually_enabled_workspace_ids=tuple(dict.fromkeys(state.manually_enabled_workspace_ids)),
+            manually_disabled_workspace_ids=tuple(dict.fromkeys(state.manually_disabled_workspace_ids)),
+            updated_at=state.updated_at,
+        )
diff --git a/api/tests/unit_tests/commands/test_generate_swagger_markdown_docs.py b/api/tests/unit_tests/commands/test_generate_swagger_markdown_docs.py
index cff6695e414..9231e274d5c 100644
--- a/api/tests/unit_tests/commands/test_generate_swagger_markdown_docs.py
+++ b/api/tests/unit_tests/commands/test_generate_swagger_markdown_docs.py
@@ -238,6 +238,42 @@ def test_patch_union_schema_markdown_fills_regular_schema_union_property(tmp_pat
     assert "| value | string<br>integer<br>number<br>boolean |  | No |" in patched
 
 
+def test_patch_union_schema_markdown_preserves_nullable_enum_values(tmp_path: Path):
+    module = _load_generate_swagger_markdown_docs_module()
+    spec_path = tmp_path / "console-openapi.json"
+    spec_path.write_text(
+        json.dumps(
+            {
+                "components": {
+                    "schemas": {
+                        "StepByStepTourStatePatchPayload": {
+                            "properties": {
+                                "task_id": {
+                                    "anyOf": [
+                                        {"enum": ["home", "studio"], "type": "string"},
+                                        {"type": "null"},
+                                    ],
+                                },
+                            },
+                        },
+                    },
+                }
+            }
+        ),
+        encoding="utf-8",
+    )
+    markdown = """#### StepByStepTourStatePatchPayload
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| task_id | string | Task ID | No |
+"""
+
+    patched = module._patch_union_schema_markdown(markdown, spec_path)
+
+    assert '| task_id | string, <br>**Available values:** "home", "studio" | Task ID | No |' in patched
+
+
 def test_patch_union_schema_markdown_fills_array_item_union_property(tmp_path: Path):
     module = _load_generate_swagger_markdown_docs_module()
     spec_path = tmp_path / "console-openapi.json"
diff --git a/api/tests/unit_tests/controllers/console/test_notification.py b/api/tests/unit_tests/controllers/console/test_notification.py
new file mode 100644
index 00000000000..48843d1af8a
--- /dev/null
+++ b/api/tests/unit_tests/controllers/console/test_notification.py
@@ -0,0 +1,77 @@
+from inspect import unwrap
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+from controllers.console.notification import (
+    DismissNotificationPayload,
+    NotificationApi,
+    NotificationDismissApi,
+)
+from machinery.context import RequestContext
+from services.entities.notification_entities import NotificationItem, NotificationResult
+
+
+def _request_context() -> RequestContext:
+    return RequestContext(
+        request_id="request-1",
+        trace_id="trace-1",
+        account_id="account-1",
+        active_workspace_id="workspace-1",
+    )
+
+
+def test_get_notification_delegates_and_serializes_result() -> None:
+    service = Mock()
+    service.get_active.return_value = NotificationResult(
+        should_show=True,
+        notifications=(
+            NotificationItem(
+                notification_id="notification-1",
+                frequency="once",
+                lang="en-US",
+                title="Title",
+                subtitle="Subtitle",
+                body="Body",
+                title_pic_url="https://example.com/title.png",
+            ),
+        ),
+    )
+    services = SimpleNamespace(notifications=service)
+    api = NotificationApi()
+    method = unwrap(api.get)
+    context = _request_context()
+
+    with patch("controllers.console.notification.application_services", return_value=services):
+        result, status = method(api, context)
+
+    assert status == 200
+    assert result == {
+        "should_show": True,
+        "notifications": [
+            {
+                "notification_id": "notification-1",
+                "frequency": "once",
+                "lang": "en-US",
+                "title": "Title",
+                "subtitle": "Subtitle",
+                "body": "Body",
+                "title_pic_url": "https://example.com/title.png",
+            }
+        ],
+    }
+    service.get_active.assert_called_once_with(context)
+
+
+def test_dismiss_notification_delegates_with_stable_account_context() -> None:
+    service = Mock()
+    services = SimpleNamespace(notifications=service)
+    api = NotificationDismissApi()
+    method = unwrap(api.post)
+    context = _request_context()
+
+    with patch("controllers.console.notification.application_services", return_value=services):
+        result, status = method(api, DismissNotificationPayload(notification_id="notification-1"), context)
+
+    assert status == 200
+    assert result == {"result": "success"}
+    service.dismiss.assert_called_once_with(context, "notification-1")
diff --git a/api/tests/unit_tests/controllers/console/test_onboarding.py b/api/tests/unit_tests/controllers/console/test_onboarding.py
index 8d613f7c202..90a9521a2fa 100644
--- a/api/tests/unit_tests/controllers/console/test_onboarding.py
+++ b/api/tests/unit_tests/controllers/console/test_onboarding.py
@@ -2,47 +2,48 @@ from __future__ import annotations
 
 from datetime import UTC, datetime
 from inspect import unwrap
-from unittest.mock import Mock
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
 
 import pytest
-from flask import Flask
 from pydantic import ValidationError
 
 from controllers.console.onboarding import (
     StepByStepTourStateApi,
     StepByStepTourStatePatchPayload,
+    StepByStepTourStateResponse,
 )
-from extensions.ext_database import db
-from models.account import Account, AccountStatus
-from services.step_by_step_tour_service import StepByStepTourService
+from machinery.context import RequestContext
+from services.entities.onboarding_entities import StepByStepTourPatch, StepByStepTourResult
 
 
-def _account() -> Account:
-    account = Account(name="User", email="user@example.com", status=AccountStatus.ACTIVE)
-    account.id = "account-1"
-    return account
+def _request_context() -> RequestContext:
+    return RequestContext(
+        request_id="request-1",
+        trace_id="trace-1",
+        account_id="account-1",
+        active_workspace_id="workspace-1",
+    )
 
 
-def _state_response() -> dict[str, object]:
-    return {
-        "first_workspace_id": "workspace-1",
-        "skipped": False,
-        "completed_task_ids": ["home"],
-        "manually_enabled_workspace_ids": [],
-        "manually_disabled_workspace_ids": [],
-        "updated_at": datetime(2026, 6, 28, tzinfo=UTC),
-    }
+def _state_result() -> StepByStepTourResult:
+    return StepByStepTourResult(
+        first_workspace_id="workspace-1",
+        completed_task_ids=("home",),
+        updated_at=datetime(2026, 6, 28, tzinfo=UTC),
+    )
 
 
-def test_get_step_by_step_tour_state(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
-    get_state = Mock(return_value=_state_response())
-    monkeypatch.setattr(StepByStepTourService, "get_state", get_state)
-
+def test_get_step_by_step_tour_state_delegates_with_request_context() -> None:
+    service = Mock()
+    service.get_state.return_value = _state_result()
+    services = SimpleNamespace(step_by_step_tour=service)
     api = StepByStepTourStateApi()
     method = unwrap(api.get)
+    context = _request_context()
 
-    with app.test_request_context("/console/api/onboarding/step-by-step-tour/state", method="GET"):
-        result = method(api, "workspace-1", _account())
+    with patch("controllers.console.onboarding.application_services", return_value=services):
+        result = method(api, context)
 
     assert result == {
         "first_workspace_id": "workspace-1",
@@ -52,35 +53,26 @@ def test_get_step_by_step_tour_state(app: Flask, monkeypatch: pytest.MonkeyPatch
         "manually_disabled_workspace_ids": [],
         "updated_at": "2026-06-28T00:00:00Z",
     }
-    get_state.assert_called_once()
-    assert get_state.call_args.kwargs["current_tenant_id"] == "workspace-1"
-    assert get_state.call_args.kwargs["session"] is db.session
+    service.get_state.assert_called_once_with(context)
 
 
-def test_patch_step_by_step_tour_state_passes_action_payload(
-    app: Flask,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    patch_state = Mock(return_value=_state_response())
-    monkeypatch.setattr(StepByStepTourService, "patch_state", patch_state)
-
+def test_patch_step_by_step_tour_state_maps_transport_payload_to_command() -> None:
+    service = Mock()
+    service.patch_state.return_value = _state_result()
+    services = SimpleNamespace(step_by_step_tour=service)
     api = StepByStepTourStateApi()
     method = unwrap(api.patch)
-    payload = {"action": "complete_task", "task_id": "studio"}
+    context = _request_context()
+    payload = StepByStepTourStatePatchPayload.model_validate({"action": "complete_task", "task_id": "studio"})
 
-    req_data = StepByStepTourStatePatchPayload.model_validate(payload)
-    with app.test_request_context(
-        "/console/api/onboarding/step-by-step-tour/state",
-        method="PATCH",
-        json=payload,
-    ):
-        result = method(api, req_data, "workspace-1", _account())
+    with patch("controllers.console.onboarding.application_services", return_value=services):
+        result = method(api, payload, context)
 
     assert result["completed_task_ids"] == ["home"]
-    patch_state.assert_called_once()
-    assert patch_state.call_args.kwargs["current_tenant_id"] == "workspace-1"
-    assert patch_state.call_args.kwargs["patch"] == payload
-    assert patch_state.call_args.kwargs["session"] is db.session
+    service.patch_state.assert_called_once_with(
+        context,
+        StepByStepTourPatch(action="complete_task", task_id="studio"),
+    )
 
 
 def test_patch_payload_rejects_non_action_fields() -> None:
@@ -96,3 +88,21 @@ def test_patch_payload_rejects_task_id_without_task_action() -> None:
 def test_patch_payload_requires_action() -> None:
     with pytest.raises(ValidationError):
         StepByStepTourStatePatchPayload.model_validate({"task_id": "home"})
+
+
+def test_step_by_step_tour_schemas_preserve_enum_values() -> None:
+    patch_schema = StepByStepTourStatePatchPayload.model_json_schema()
+    action_schema = patch_schema["properties"]["action"]
+    task_id_schema = patch_schema["properties"]["task_id"]
+    task_id_values = next(candidate["enum"] for candidate in task_id_schema["anyOf"] if "enum" in candidate)
+    response_schema = StepByStepTourStateResponse.model_json_schema()
+
+    assert set(action_schema["enum"]) == {
+        "skip",
+        "complete_task",
+        "uncomplete_task",
+        "enable_current_workspace",
+        "disable_current_workspace",
+    }
+    assert set(task_id_values) == {"home", "studio", "knowledge", "integration"}
+    assert set(response_schema["properties"]["completed_task_ids"]["items"]["enum"]) == set(task_id_values)
diff --git a/api/tests/unit_tests/extensions/test_ext_application_services.py b/api/tests/unit_tests/extensions/test_ext_application_services.py
index ee3d2125204..191f13cf7e3 100644
--- a/api/tests/unit_tests/extensions/test_ext_application_services.py
+++ b/api/tests/unit_tests/extensions/test_ext_application_services.py
@@ -382,6 +382,8 @@ def test_build_application_services_wires_account_profile_repository(
     assert email_registration._registration._session_factory is sqlite_session_factory
     assert services.accounts.education._accounts is accounts
     assert services.accounts.deletion._accounts is accounts
+    assert services.notifications._accounts is accounts
+    assert services.step_by_step_tour._accounts is accounts
     assert services.accounts.deletion._memberships is services.workspace_queries._workspaces
     integrations = services.accounts.integrations._integrations
     assert isinstance(integrations, SQLAlchemyAccountIntegrationRepository)
diff --git a/api/tests/unit_tests/repositories/test_step_by_step_tour_repository.py b/api/tests/unit_tests/repositories/test_step_by_step_tour_repository.py
new file mode 100644
index 00000000000..a6439f58bc5
--- /dev/null
+++ b/api/tests/unit_tests/repositories/test_step_by_step_tour_repository.py
@@ -0,0 +1,171 @@
+from contextlib import nullcontext
+from dataclasses import replace
+from datetime import datetime
+from typing import cast
+from unittest.mock import MagicMock, Mock
+
+import pytest
+from sqlalchemy.exc import IntegrityError, OperationalError
+from sqlalchemy.orm import Session, sessionmaker
+
+from models.onboarding import AccountStepByStepTourState
+from repositories.step_by_step_tour_repository import (
+    SQLAlchemyStepByStepTourStateRepository,
+    _is_retryable_mysql_lock_error,
+)
+
+
+class _ErrnoOnlyError(Exception):
+    def __init__(self, errno: int | str) -> None:
+        super().__init__()
+        self.errno = errno
+
+
+def test_mutate_creates_and_updates_state_in_repository_owned_transaction(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    repository = SQLAlchemyStepByStepTourStateRepository(sqlite_session_factory)
+
+    saved = repository.mutate(
+        "account-1",
+        lambda state: replace(state, completed_task_ids=("home",)),
+    )
+    reloaded = repository.get("account-1")
+
+    assert saved.first_workspace_id is None
+    assert saved.completed_task_ids == ("home",)
+    assert saved.updated_at is not None
+    assert reloaded == saved
+
+
+def test_initialize_creates_state_with_first_workspace_atomically(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    repository = SQLAlchemyStepByStepTourStateRepository(sqlite_session_factory)
+
+    result = repository.initialize("account-1", "workspace-1")
+
+    assert result.first_workspace_id == "workspace-1"
+    assert repository.get("account-1") == result
+
+
+def test_initialize_claims_empty_state_once_without_overwriting_winner(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    repository = SQLAlchemyStepByStepTourStateRepository(sqlite_session_factory)
+    with sqlite_session_factory() as session:
+        session.add(AccountStepByStepTourState(account_id="account-1"))
+        session.commit()
+
+    first = repository.initialize("account-1", "workspace-1")
+    second = repository.initialize("account-1", "workspace-2")
+
+    assert first.first_workspace_id == "workspace-1"
+    assert second.first_workspace_id == "workspace-1"
+
+
+def test_mutate_cannot_clear_or_overwrite_first_workspace(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    repository = SQLAlchemyStepByStepTourStateRepository(sqlite_session_factory)
+    repository.initialize("account-1", "workspace-1")
+
+    result = repository.mutate(
+        "account-1",
+        lambda state: replace(state, first_workspace_id="workspace-2", skipped=True),
+    )
+
+    assert result.first_workspace_id == "workspace-1"
+    assert result.skipped is True
+
+
+def test_sequential_mutations_replay_against_latest_state(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    repository = SQLAlchemyStepByStepTourStateRepository(sqlite_session_factory)
+
+    repository.mutate("account-1", lambda state: replace(state, completed_task_ids=("home",)))
+    result = repository.mutate(
+        "account-1",
+        lambda state: replace(state, completed_task_ids=(*state.completed_task_ids, "studio")),
+    )
+
+    assert result.completed_task_ids == ("home", "studio")
+
+
+def test_mutate_replays_after_concurrent_create_conflict() -> None:
+    concurrent_state = AccountStepByStepTourState(account_id="account-1")
+    concurrent_state.completed_task_ids = ["home"]
+    concurrent_state.updated_at = datetime(2026, 8, 13)
+    session = MagicMock(spec=Session)
+    session.execute.return_value.scalar_one_or_none.side_effect = [None, concurrent_state]
+    session.flush.side_effect = IntegrityError("insert", {}, Exception("duplicate"))
+    factory = cast(sessionmaker[Session], Mock(return_value=nullcontext(session)))
+    repository = SQLAlchemyStepByStepTourStateRepository(factory)
+
+    result = repository.mutate(
+        "account-1",
+        lambda state: replace(state, completed_task_ids=(*state.completed_task_ids, "studio")),
+    )
+
+    assert result.completed_task_ids == ("home", "studio")
+    session.rollback.assert_called_once_with()
+    initial_probe = session.execute.call_args_list[0].args[0]
+    replay_statement = session.execute.call_args_list[1].args[0]
+    assert initial_probe._for_update_arg is None
+    assert replay_statement._for_update_arg is not None
+
+
+def test_mutate_retries_mysql_deadlock_with_fresh_session() -> None:
+    concurrent_state = AccountStepByStepTourState(account_id="account-1")
+    concurrent_state.completed_task_ids = ["home"]
+    concurrent_state.updated_at = datetime(2026, 8, 13)
+
+    deadlocked_session = MagicMock(spec=Session)
+    deadlocked_session.execute.return_value.scalar_one_or_none.return_value = None
+    deadlocked_session.flush.side_effect = OperationalError(
+        "INSERT",
+        {},
+        Exception(1213, "Deadlock found when trying to get lock"),
+    )
+
+    retry_session = MagicMock(spec=Session)
+    retry_session.execute.return_value.scalar_one_or_none.side_effect = [concurrent_state, concurrent_state]
+    factory = Mock(side_effect=[nullcontext(deadlocked_session), nullcontext(retry_session)])
+    repository = SQLAlchemyStepByStepTourStateRepository(cast(sessionmaker[Session], factory))
+
+    result = repository.mutate(
+        "account-1",
+        lambda state: replace(state, completed_task_ids=(*state.completed_task_ids, "studio")),
+    )
+
+    assert result.completed_task_ids == ("home", "studio")
+    assert factory.call_count == 2
+    retry_lock_statement = retry_session.execute.call_args_list[1].args[0]
+    assert retry_lock_statement._for_update_arg is not None
+
+
+@pytest.mark.parametrize(
+    ("orig", "expected"),
+    [
+        pytest.param(_ErrnoOnlyError(1205), True, id="errno-attribute"),
+        pytest.param(Exception(1213, "deadlock"), True, id="integer-args-code"),
+        pytest.param(Exception("1213", "deadlock"), True, id="string-args-code"),
+        pytest.param(Exception(9999, "other error"), False, id="non-retryable-code"),
+        pytest.param(Exception(True), False, id="boolean-is-not-an-error-code"),
+        pytest.param(Exception(), False, id="missing-error-code"),
+    ],
+)
+def test_mysql_lock_error_detection_preserves_errno_and_args_coverage(
+    orig: BaseException,
+    expected: bool,
+) -> None:
+    exc = OperationalError("statement", {}, orig)
+
+    assert _is_retryable_mysql_lock_error(exc) is expected
+
+
+def test_get_returns_none_for_unknown_account(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    assert SQLAlchemyStepByStepTourStateRepository(sqlite_session_factory).get("missing") is None
diff --git a/api/tests/unit_tests/services/test_notification_gateway.py b/api/tests/unit_tests/services/test_notification_gateway.py
new file mode 100644
index 00000000000..9df67e7a325
--- /dev/null
+++ b/api/tests/unit_tests/services/test_notification_gateway.py
@@ -0,0 +1,63 @@
+from unittest.mock import patch
+
+from services.entities.notification_entities import NotificationContent
+from services.notification_gateway import BillingNotificationGateway
+
+
+def test_get_active_maps_billing_proto_json_contract() -> None:
+    payload = {
+        "shouldShow": True,
+        "notifications": [
+            {
+                "notificationId": "notification-1",
+                "frequency": "once",
+                "contents": {
+                    "en-US": {
+                        "lang": "en-US",
+                        "title": "Title",
+                        "subtitle": "Subtitle",
+                        "body": "Body",
+                        "titlePicUrl": "title.png",
+                    }
+                },
+            }
+        ],
+    }
+
+    with patch("services.notification_gateway.BillingService.get_account_notification", return_value=payload):
+        result = BillingNotificationGateway().get_active("account-1")
+
+    assert result.should_show is True
+    assert result.notifications[0].notification_id == "notification-1"
+    assert result.notifications[0].contents["en-US"].title_pic_url == "title.png"
+
+
+def test_get_active_omits_empty_localized_content_so_service_can_fall_back() -> None:
+    empty_localized_content: dict[str, str] = {}
+    payload = {
+        "shouldShow": True,
+        "notifications": [
+            {
+                "notificationId": "notification-1",
+                "frequency": "once",
+                "contents": {
+                    "zh-Hans": empty_localized_content,
+                    "en-US": {"lang": "en-US", "title": "Title"},
+                },
+            }
+        ],
+    }
+
+    with patch("services.notification_gateway.BillingService.get_account_notification", return_value=payload):
+        result = BillingNotificationGateway().get_active("account-1")
+
+    assert result.notifications[0].contents == {
+        "en-US": NotificationContent("en-US", "Title", "", "", ""),
+    }
+
+
+def test_dismiss_delegates_to_billing_service() -> None:
+    with patch("services.notification_gateway.BillingService.dismiss_notification") as dismiss:
+        BillingNotificationGateway().dismiss("notification-1", "account-1")
+
+    dismiss.assert_called_once_with(notification_id="notification-1", account_id="account-1")
diff --git a/api/tests/unit_tests/services/test_notification_service.py b/api/tests/unit_tests/services/test_notification_service.py
new file mode 100644
index 00000000000..3be7f08a6f7
--- /dev/null
+++ b/api/tests/unit_tests/services/test_notification_service.py
@@ -0,0 +1,138 @@
+from datetime import datetime
+from unittest.mock import Mock
+
+import pytest
+
+from machinery.context import RequestContext
+from services.account_ports import AccountRepository
+from services.entities.account_entities import AccountSnapshot
+from services.entities.notification_entities import (
+    AccountNotification,
+    AccountNotificationBatch,
+    NotificationContent,
+    NotificationItem,
+    NotificationResult,
+)
+from services.notification_service import NotificationService
+
+
+def _context() -> RequestContext:
+    return RequestContext(
+        request_id="request-1",
+        trace_id="trace-1",
+        account_id="account-1",
+        active_workspace_id="workspace-1",
+    )
+
+
+class NotificationGatewayStub:
+    def __init__(self, batch: AccountNotificationBatch) -> None:
+        self.batch = batch
+        self.get_account_ids: list[str] = []
+        self.dismissals: list[tuple[str, str]] = []
+
+    def get_active(self, account_id: str) -> AccountNotificationBatch:
+        self.get_account_ids.append(account_id)
+        return self.batch
+
+    def dismiss(self, notification_id: str, account_id: str) -> None:
+        self.dismissals.append((notification_id, account_id))
+
+
+def _account(language: str | None = "zh-Hans") -> AccountSnapshot:
+    return AccountSnapshot(
+        id="account-1",
+        name="Account",
+        email="account@example.com",
+        avatar=None,
+        is_password_set=False,
+        interface_language=language,
+        interface_theme="light",
+        timezone="UTC",
+        last_login_at=None,
+        last_login_ip=None,
+        status="active",
+        initialized_at=None,
+        created_at=datetime(2026, 1, 1),
+    )
+
+
+def _accounts(account: AccountSnapshot | None) -> Mock:
+    accounts = Mock(spec=AccountRepository)
+    accounts.get.return_value = account
+    return accounts
+
+
+def _notification(contents: dict[str, NotificationContent]) -> AccountNotification:
+    return AccountNotification(
+        notification_id="notification-1",
+        frequency="once",
+        contents=contents,
+    )
+
+
+def test_get_active_localizes_notification_for_account_language() -> None:
+    chinese = NotificationContent("zh-Hans", "标题", "副标题", "正文", "zh.png")
+    english = NotificationContent("en-US", "Title", "Subtitle", "Body", "en.png")
+    gateway = NotificationGatewayStub(
+        AccountNotificationBatch(True, (_notification({"zh-Hans": chinese, "en-US": english}),))
+    )
+    service = NotificationService(accounts=_accounts(_account()), notifications=gateway)
+
+    result = service.get_active(_context())
+
+    assert result == NotificationResult(
+        should_show=True,
+        notifications=(NotificationItem("notification-1", "once", "zh-Hans", "标题", "副标题", "正文", "zh.png"),),
+    )
+    assert gateway.get_account_ids == ["account-1"]
+
+
+def test_get_active_falls_back_to_english() -> None:
+    english = NotificationContent("en-US", "Title", "Subtitle", "Body", "en.png")
+    gateway = NotificationGatewayStub(AccountNotificationBatch(True, (_notification({"en-US": english}),)))
+    service = NotificationService(accounts=_accounts(_account("fr-FR")), notifications=gateway)
+
+    result = service.get_active(_context())
+
+    assert result.notifications[0].lang == "en-US"
+    assert result.notifications[0].title == "Title"
+
+
+def test_get_active_skips_account_query_when_gateway_says_not_to_show() -> None:
+    accounts = _accounts(None)
+    service = NotificationService(
+        accounts=accounts,
+        notifications=NotificationGatewayStub(AccountNotificationBatch(False, ())),
+    )
+
+    result = service.get_active(_context())
+
+    assert result == NotificationResult(False, ())
+    accounts.get.assert_not_called()
+
+
+def test_get_active_uses_empty_content_when_notification_has_no_translations() -> None:
+    gateway = NotificationGatewayStub(AccountNotificationBatch(True, (_notification({}),)))
+    service = NotificationService(accounts=_accounts(_account(None)), notifications=gateway)
+
+    result = service.get_active(_context())
+
+    assert result.notifications == (NotificationItem("notification-1", "once", "en-US", "", "", "", ""),)
+
+
+def test_get_active_rejects_unknown_admitted_account() -> None:
+    gateway = NotificationGatewayStub(AccountNotificationBatch(True, (_notification({}),)))
+    service = NotificationService(accounts=_accounts(None), notifications=gateway)
+
+    with pytest.raises(RuntimeError, match="unknown account"):
+        service.get_active(_context())
+
+
+def test_dismiss_delegates_identifiers_to_gateway() -> None:
+    gateway = NotificationGatewayStub(AccountNotificationBatch(False, ()))
+    service = NotificationService(accounts=_accounts(_account()), notifications=gateway)
+
+    service.dismiss(_context(), "notification-1")
+
+    assert gateway.dismissals == [("notification-1", "account-1")]
diff --git a/api/tests/unit_tests/services/test_step_by_step_tour_service.py b/api/tests/unit_tests/services/test_step_by_step_tour_service.py
index 7a99fa61444..40017bb7798 100644
--- a/api/tests/unit_tests/services/test_step_by_step_tour_service.py
+++ b/api/tests/unit_tests/services/test_step_by_step_tour_service.py
@@ -1,230 +1,213 @@
 from __future__ import annotations
 
-from datetime import UTC, datetime
+from collections.abc import Callable
+from dataclasses import replace
+from datetime import datetime
+from unittest.mock import Mock
 
 import pytest
-from sqlalchemy import event, select
-from sqlalchemy.orm import Session, sessionmaker
 
-from enums import DeploymentEdition
-from models.account import Account, AccountStatus
-from models.onboarding import AccountStepByStepTourState
+from machinery.context import RequestContext
+from services.account_ports import AccountRepository
+from services.entities.account_entities import AccountSnapshot
+from services.entities.onboarding_entities import StepByStepTourPatch, StepByStepTourResult, StepByStepTourState
 from services.step_by_step_tour_service import StepByStepTourService
-from tests.unit_tests.config_override import apply_config_overrides
 
 
-def _account(*, initialized_at: datetime | None = None, created_at: datetime | None = None) -> Account:
-    account = Account(name="User", email="user@example.com", status=AccountStatus.ACTIVE)
-    account.id = "account-1"
-    account.initialized_at = initialized_at
-    account.created_at = created_at or datetime(2026, 6, 28)
-    return account
-
-
-def _state() -> AccountStepByStepTourState:
-    state = AccountStepByStepTourState(account_id="account-1")
-    state.updated_at = datetime(2026, 6, 28, tzinfo=UTC)
-    return state
-
-
-def _persist_state(session: Session, state: AccountStepByStepTourState) -> None:
-    session.add(state)
-    session.commit()
-
-
-def _load_state(session: Session) -> AccountStepByStepTourState | None:
-    return session.scalar(
-        select(AccountStepByStepTourState).where(AccountStepByStepTourState.account_id == "account-1")
+def _context(*, workspace_id: str | None = "workspace-1") -> RequestContext:
+    return RequestContext(
+        request_id="request-1",
+        trace_id="trace-1",
+        account_id="account-1",
+        active_workspace_id=workspace_id,
     )
 
 
-def _set_tour_config(monkeypatch: pytest.MonkeyPatch, *, enabled: bool, rollout_started_at: datetime | None) -> None:
-    apply_config_overrides(
-        monkeypatch,
-        ENABLE_STEP_BY_STEP_TOUR=enabled,
-        STEP_BY_STEP_TOUR_ROLLOUT_STARTED_AT=rollout_started_at,
+class StateRepositoryStub:
+    def __init__(self, state: StepByStepTourState | None = None) -> None:
+        self.state = state
+        self.get_account_ids: list[str] = []
+        self.initialize_calls: list[tuple[str, str]] = []
+        self.mutation_account_ids: list[str] = []
+
+    def get(self, account_id: str) -> StepByStepTourState | None:
+        self.get_account_ids.append(account_id)
+        return self.state
+
+    def initialize(self, account_id: str, first_workspace_id: str) -> StepByStepTourState:
+        self.initialize_calls.append((account_id, first_workspace_id))
+        if self.state is None:
+            self.state = StepByStepTourState(account_id=account_id, first_workspace_id=first_workspace_id)
+        elif self.state.first_workspace_id is None:
+            self.state = replace(self.state, first_workspace_id=first_workspace_id)
+        return self.state
+
+    def mutate(
+        self,
+        account_id: str,
+        mutation: Callable[[StepByStepTourState], StepByStepTourState],
+    ) -> StepByStepTourState:
+        self.mutation_account_ids.append(account_id)
+        if self.state is None:
+            self.state = StepByStepTourState(account_id=account_id)
+        self.state = mutation(self.state)
+        return self.state
+
+
+def _account(*, started_at: datetime = datetime(2026, 6, 28)) -> AccountSnapshot:
+    return AccountSnapshot(
+        id="account-1",
+        name="Account",
+        email="account@example.com",
+        avatar=None,
+        is_password_set=False,
+        interface_language="en-US",
+        interface_theme="light",
+        timezone="UTC",
+        last_login_at=None,
+        last_login_ip=None,
+        status="active",
+        initialized_at=started_at,
+        created_at=started_at,
     )
 
 
-def test_get_state_creates_state_and_records_first_workspace_for_eligible_account(
-    monkeypatch: pytest.MonkeyPatch,
-    sqlite_session: Session,
-    sqlite_session_factory: sessionmaker[Session],
-) -> None:
-    _set_tour_config(monkeypatch, enabled=True, rollout_started_at=datetime(2026, 6, 1))
+def _accounts(account: AccountSnapshot | None) -> Mock:
+    accounts = Mock(spec=AccountRepository)
+    accounts.get.return_value = account
+    return accounts
 
-    result = StepByStepTourService.get_state(
-        account=_account(initialized_at=datetime(2026, 6, 28)),
-        current_tenant_id="workspace-1",
-        session=sqlite_session,
+
+def _service(
+    *,
+    states: StateRepositoryStub,
+    account: AccountSnapshot | None = None,
+    enabled: bool = True,
+    rollout_started_at: datetime | None = datetime(2026, 6, 1),
+) -> StepByStepTourService:
+    return StepByStepTourService(
+        accounts=_accounts(account or _account()),
+        states=states,
+        enabled=enabled,
+        rollout_started_at=rollout_started_at,
     )
 
-    assert result["first_workspace_id"] == "workspace-1"
-    assert result["completed_task_ids"] == []
-    with sqlite_session_factory() as observer:
-        persisted = _load_state(observer)
-        assert persisted is not None
-        assert persisted.account_id == "account-1"
-        assert persisted.first_workspace_id == "workspace-1"
+
+def test_get_state_creates_state_and_records_first_workspace_for_eligible_account() -> None:
+    states = StateRepositoryStub()
+
+    result = _service(states=states).get_state(_context())
+
+    assert result.first_workspace_id == "workspace-1"
+    assert states.get_account_ids == []
+    assert states.initialize_calls == [("account-1", "workspace-1")]
+    assert states.mutation_account_ids == []
 
 
-def test_is_eligible_does_not_depend_on_cloud_edition(monkeypatch: pytest.MonkeyPatch) -> None:
-    _set_tour_config(monkeypatch, enabled=True, rollout_started_at=datetime(2026, 6, 1))
-    apply_config_overrides(monkeypatch, DEPLOYMENT_EDITION=DeploymentEdition.COMMUNITY)
+def test_get_state_returns_existing_state_without_rewriting_first_workspace() -> None:
+    state = StepByStepTourState(account_id="account-1", first_workspace_id="workspace-original")
+    states = StateRepositoryStub(state)
 
-    result = StepByStepTourService.is_eligible(_account(initialized_at=datetime(2026, 6, 28)))
+    result = _service(states=states).get_state(_context(workspace_id="workspace-current"))
 
-    assert result is True
+    assert result.first_workspace_id == "workspace-original"
+    assert states.initialize_calls == [("account-1", "workspace-current")]
+    assert states.mutation_account_ids == []
 
 
-def test_get_state_does_not_create_state_for_ineligible_account_without_existing_state(
-    monkeypatch: pytest.MonkeyPatch,
-    sqlite_session: Session,
-    sqlite_session_factory: sessionmaker[Session],
-) -> None:
-    _set_tour_config(monkeypatch, enabled=True, rollout_started_at=datetime(2026, 6, 1))
+def test_get_state_does_not_create_state_for_ineligible_account() -> None:
+    states = StateRepositoryStub()
+    service = _service(states=states, account=_account(started_at=datetime(2026, 5, 31)))
 
-    result = StepByStepTourService.get_state(
-        account=_account(initialized_at=datetime(2026, 5, 31)),
-        current_tenant_id="workspace-1",
-        session=sqlite_session,
+    result = service.get_state(_context())
+
+    assert result == StepByStepTourResult()
+    assert states.get_account_ids == ["account-1"]
+    assert states.mutation_account_ids == []
+
+
+def test_get_state_does_not_create_state_when_tour_is_disabled() -> None:
+    states = StateRepositoryStub()
+
+    result = _service(states=states, enabled=False).get_state(_context())
+
+    assert result == StepByStepTourResult()
+    assert states.get_account_ids == ["account-1"]
+
+
+def test_patch_state_persists_even_when_tour_is_disabled() -> None:
+    states = StateRepositoryStub()
+    service = _service(states=states, enabled=False)
+
+    result = service.patch_state(_context(workspace_id="workspace-2"), StepByStepTourPatch("enable_current_workspace"))
+
+    assert result.manually_enabled_workspace_ids == ("workspace-2",)
+    assert states.mutation_account_ids == ["account-1"]
+
+
+def test_patch_state_skip_removes_current_workspace_enable() -> None:
+    states = StateRepositoryStub(
+        StepByStepTourState(
+            account_id="account-1",
+            manually_enabled_workspace_ids=("workspace-1", "workspace-2"),
+        )
     )
 
-    assert result == {
-        "first_workspace_id": None,
-        "skipped": False,
-        "completed_task_ids": [],
-        "manually_enabled_workspace_ids": [],
-        "manually_disabled_workspace_ids": [],
-        "updated_at": None,
-    }
-    with sqlite_session_factory() as observer:
-        assert _load_state(observer) is None
+    result = _service(states=states).patch_state(_context(), StepByStepTourPatch("skip"))
+
+    assert result.skipped is True
+    assert result.manually_enabled_workspace_ids == ("workspace-2",)
 
 
-def test_patch_state_persists_even_when_account_is_not_eligible(
-    monkeypatch: pytest.MonkeyPatch,
-    sqlite_session: Session,
-    sqlite_session_factory: sessionmaker[Session],
-) -> None:
-    _set_tour_config(monkeypatch, enabled=False, rollout_started_at=datetime(2026, 6, 1))
-
-    result = StepByStepTourService.patch_state(
-        account=_account(initialized_at=datetime(2026, 6, 28)),
-        current_tenant_id="workspace-2",
-        patch={"action": "enable_current_workspace"},
-        session=sqlite_session,
+def test_patch_state_disable_moves_current_workspace_to_disabled() -> None:
+    states = StateRepositoryStub(
+        StepByStepTourState(
+            account_id="account-1",
+            manually_enabled_workspace_ids=("workspace-1", "workspace-2"),
+        )
     )
 
-    assert result["skipped"] is False
-    assert result["manually_enabled_workspace_ids"] == ["workspace-2"]
-    assert result["manually_disabled_workspace_ids"] == []
-    with sqlite_session_factory() as observer:
-        persisted = _load_state(observer)
-        assert persisted is not None
-        assert persisted.manually_enabled_workspace_ids == ["workspace-2"]
-
-
-def test_patch_state_skip_action_sets_skipped_and_removes_current_workspace_enable(
-    monkeypatch: pytest.MonkeyPatch,
-    sqlite_session: Session,
-) -> None:
-    _set_tour_config(monkeypatch, enabled=False, rollout_started_at=datetime(2026, 6, 1))
-    state = _state()
-    state.manually_enabled_workspace_ids = ["workspace-1", "workspace-2"]
-    _persist_state(sqlite_session, state)
-
-    result = StepByStepTourService.patch_state(
-        account=_account(initialized_at=datetime(2026, 6, 28)),
-        current_tenant_id="workspace-1",
-        patch={"action": "skip"},
-        session=sqlite_session,
+    result = _service(states=states).patch_state(
+        _context(),
+        StepByStepTourPatch("disable_current_workspace"),
     )
 
-    assert result["skipped"] is True
-    assert result["manually_enabled_workspace_ids"] == ["workspace-2"]
-    assert result["manually_disabled_workspace_ids"] == []
-    assert _load_state(sqlite_session) is state
+    assert result.manually_enabled_workspace_ids == ("workspace-2",)
+    assert result.manually_disabled_workspace_ids == ("workspace-1",)
 
 
-def test_patch_state_disable_action_moves_current_workspace_to_disabled(
-    monkeypatch: pytest.MonkeyPatch,
-    sqlite_session: Session,
-) -> None:
-    _set_tour_config(monkeypatch, enabled=False, rollout_started_at=datetime(2026, 6, 1))
-    state = _state()
-    state.manually_enabled_workspace_ids = ["workspace-1", "workspace-2"]
-    _persist_state(sqlite_session, state)
+def test_patch_state_complete_and_uncomplete_task() -> None:
+    states = StateRepositoryStub(StepByStepTourState(account_id="account-1", completed_task_ids=("home",)))
+    service = _service(states=states)
 
-    result = StepByStepTourService.patch_state(
-        account=_account(initialized_at=datetime(2026, 6, 28)),
-        current_tenant_id="workspace-1",
-        patch={"action": "disable_current_workspace"},
-        session=sqlite_session,
+    service.patch_state(_context(), StepByStepTourPatch("complete_task", "studio"))
+    result = service.patch_state(_context(), StepByStepTourPatch("uncomplete_task", "home"))
+
+    assert result.completed_task_ids == ("studio",)
+
+
+def test_rejects_unsupported_task_id() -> None:
+    with pytest.raises(ValueError, match="Unsupported task_id"):
+        StepByStepTourService._require_task_id("unknown")
+
+
+def test_rejects_missing_workspace_before_using_state_repository() -> None:
+    states = StateRepositoryStub()
+
+    with pytest.raises(RuntimeError, match="did not resolve an active workspace"):
+        _service(states=states).patch_state(_context(workspace_id=None), StepByStepTourPatch("skip"))
+
+    assert states.mutation_account_ids == []
+
+
+def test_get_state_rejects_unknown_admitted_account() -> None:
+    states = StateRepositoryStub()
+    service = StepByStepTourService(
+        accounts=_accounts(None),
+        states=states,
+        enabled=True,
+        rollout_started_at=datetime(2026, 6, 1),
     )
 
-    assert result["manually_enabled_workspace_ids"] == ["workspace-2"]
-    assert result["manually_disabled_workspace_ids"] == ["workspace-1"]
-    assert _load_state(sqlite_session) is state
-
-
-def test_patch_state_complete_and_uncomplete_task(
-    monkeypatch: pytest.MonkeyPatch,
-    sqlite_session: Session,
-) -> None:
-    _set_tour_config(monkeypatch, enabled=False, rollout_started_at=datetime(2026, 6, 1))
-    state = _state()
-    state.completed_task_ids = ["home"]
-    _persist_state(sqlite_session, state)
-
-    StepByStepTourService.patch_state(
-        account=_account(initialized_at=datetime(2026, 6, 28)),
-        current_tenant_id="workspace-1",
-        patch={"action": "complete_task", "task_id": "studio"},
-        session=sqlite_session,
-    )
-    result = StepByStepTourService.patch_state(
-        account=_account(initialized_at=datetime(2026, 6, 28)),
-        current_tenant_id="workspace-1",
-        patch={"action": "uncomplete_task", "task_id": "home"},
-        session=sqlite_session,
-    )
-
-    assert result["completed_task_ids"] == ["studio"]
-
-
-def test_patch_state_recovers_when_concurrent_request_created_state(
-    monkeypatch: pytest.MonkeyPatch,
-    sqlite_session: Session,
-    sqlite_session_factory: sessionmaker[Session],
-) -> None:
-    _set_tour_config(monkeypatch, enabled=False, rollout_started_at=datetime(2026, 6, 1))
-    existing_state = _state()
-    existing_state.manually_enabled_workspace_ids = ["workspace-1"]
-    lifecycle_events: list[str] = []
-
-    @event.listens_for(sqlite_session, "before_flush", once=True)
-    def add_conflicting_pending_state(session: Session, _flush_context, _instances) -> None:
-        lifecycle_events.append("before_flush")
-        session.add(AccountStepByStepTourState(account_id="account-1"))
-
-    @event.listens_for(sqlite_session, "after_soft_rollback", once=True)
-    def persist_winning_request(_session: Session, _previous_transaction) -> None:
-        lifecycle_events.append("after_soft_rollback")
-        with sqlite_session_factory() as winner:
-            winner.add(existing_state)
-            winner.commit()
-
-    result = StepByStepTourService.patch_state(
-        account=_account(initialized_at=datetime(2026, 6, 28)),
-        current_tenant_id="workspace-2",
-        patch={"action": "enable_current_workspace"},
-        session=sqlite_session,
-    )
-
-    assert result["manually_enabled_workspace_ids"] == ["workspace-1", "workspace-2"]
-    assert lifecycle_events == ["before_flush", "after_soft_rollback"]
-    with sqlite_session_factory() as observer:
-        persisted = _load_state(observer)
-        assert persisted is not None
-        assert persisted.manually_enabled_workspace_ids == ["workspace-1", "workspace-2"]
+    with pytest.raises(RuntimeError, match="unknown account"):
+        service.get_state(_context())