feat: step by step backend

api/configs/feature/__init__.py

@@ -1,4 +1,4 @@
-from datetime import timedelta
+from datetime import datetime, timedelta
 from enum import StrEnum
 from typing import Literal
 
@@ -1094,6 +1094,16 @@ class HomepageConfig(BaseSettings):
         default=True,
     )
 
+    ENABLE_STEP_BY_STEP_TOUR: bool = Field(
+        description="Enable account-level Step-by-step Tour eligibility checks",
+        default=False,
+    )
+
+    STEP_BY_STEP_TOUR_ROLLOUT_STARTED_AT: datetime | None = Field(
+        description="UTC timestamp after which newly initialized accounts are eligible for Step-by-step Tour",
+        default=None,
+    )
+
 
 class RagEtlConfig(BaseSettings):
     """

api/controllers/console/__init__.py

@@ -39,6 +39,7 @@ from . import (
     human_input_form,
     init_validate,
     notification,
+    onboarding,
     ping,
     setup,
     spec,
@@ -204,6 +205,7 @@ __all__ = [
     "notification",
     "oauth",
     "oauth_server",
+    "onboarding",
     "ops_trace",
     "parameter",
     "ping",

api/controllers/console/onboarding.py

@@ -0,0 +1,107 @@
+"""Console onboarding APIs.
+
+This module keeps Step-by-step Tour persistence account-scoped. Workspace IDs
+are accepted only as presentation overrides; UI-only state such as minimized
+panels or the currently active task stays on the frontend. PATCH requests are
+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 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 . import console_ns
+from .wraps import account_initialization_required, 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):
+    action: StepByStepTourAction = Field(description="State update action")
+    task_id: StepByStepTourTaskId | None = Field(default=None, description="Task ID for task actions")
+
+    model_config = ConfigDict(extra="forbid")
+
+    @model_validator(mode="after")
+    def validate_patch_shape(self) -> "StepByStepTourStatePatchPayload":
+        task_actions = {"complete_task", "uncomplete_task"}
+        if self.action in task_actions and self.task_id is None:
+            raise ValueError("task_id is required for task actions")
+        if self.action not in task_actions and self.task_id is not None:
+            raise ValueError("task_id is only supported for task actions")
+
+        return self
+
+
+class StepByStepTourStateResponse(ResponseModel):
+    eligible: bool
+    first_workspace_id: str | None = None
+    skipped: bool = False
+    completed_task_ids: list[StepByStepTourTaskId] = Field(default_factory=list)
+    manually_enabled_workspace_ids: list[str] = Field(default_factory=list)
+    manually_disabled_workspace_ids: list[str] = Field(default_factory=list)
+    updated_at: datetime | None = None
+
+
+register_schema_models(console_ns, StepByStepTourStatePatchPayload)
+register_response_schema_models(console_ns, StepByStepTourStateResponse)
+
+
+@console_ns.route("/onboarding/step-by-step-tour/state")
+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):
+        return dump_response(
+            StepByStepTourStateResponse,
+            StepByStepTourService.get_state(
+                account=current_user,
+                current_tenant_id=current_tenant_id,
+                session=db.session,
+            ),
+        )
+
+    @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
+    def patch(self, current_tenant_id: str, current_user: Account):
+        payload = StepByStepTourStatePatchPayload.model_validate(console_ns.payload or {})
+        patch = cast(StepByStepTourPatch, payload.model_dump(exclude_unset=True, exclude_none=True))
+        return dump_response(
+            StepByStepTourStateResponse,
+            StepByStepTourService.patch_state(
+                account=current_user,
+                current_tenant_id=current_tenant_id,
+                patch=patch,
+                session=db.session,
+            ),
+        )

api/migrations/versions/2026_06_29_1200-e4f5a6b7c8d9_add_step_by_step_tour_state.py

@@ -0,0 +1,39 @@
+"""add step by step tour state
+
+Revision ID: e4f5a6b7c8d9
+Revises: d9e8f7a6b5c4
+Create Date: 2026-06-29 12:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+import models
+
+# revision identifiers, used by Alembic.
+revision = "e4f5a6b7c8d9"
+down_revision = "d9e8f7a6b5c4"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "account_step_by_step_tour_states",
+        sa.Column("id", models.types.StringUUID(), nullable=False),
+        sa.Column("account_id", models.types.StringUUID(), nullable=False),
+        sa.Column("first_workspace_id", models.types.StringUUID(), nullable=True),
+        sa.Column("skipped", sa.Boolean(), server_default=sa.text("false"), nullable=False),
+        sa.Column("completed_task_ids", models.types.AdjustedJSON(), nullable=False),
+        sa.Column("manually_enabled_workspace_ids", models.types.AdjustedJSON(), nullable=False),
+        sa.Column("manually_disabled_workspace_ids", models.types.AdjustedJSON(), nullable=False),
+        sa.Column("created_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
+        sa.PrimaryKeyConstraint("id", name="account_step_by_step_tour_state_pkey"),
+        sa.UniqueConstraint("account_id", name="account_step_by_step_tour_state_account_id_key"),
+    )
+
+
+def downgrade():
+    op.drop_table("account_step_by_step_tour_states")

api/models/__init__.py

@@ -99,6 +99,7 @@ from .model import (
     UploadFile,
 )
 from .oauth import DatasourceOauthParamConfig, DatasourceProvider, OAuthAccessToken
+from .onboarding import AccountStepByStepTourState
 from .provider import (
     LoadBalancingModelConfig,
     Provider,
@@ -152,6 +153,7 @@ __all__ = [
     "Account",
     "AccountIntegrate",
     "AccountStatus",
+    "AccountStepByStepTourState",
     "AccountTrialAppRecord",
     "Agent",
     "AgentConfigRevision",

api/models/onboarding.py

@@ -0,0 +1,59 @@
+"""Account-level onboarding state models."""
+
+from datetime import datetime
+
+import sqlalchemy as sa
+from sqlalchemy import DateTime, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from .base import TypeBase, gen_uuidv7_string
+from .types import AdjustedJSON, StringUUID
+
+
+class AccountStepByStepTourState(TypeBase):
+    """Persistent account-level Step-by-step Tour state.
+
+    The tour is account-owned, with workspace IDs stored only as presentation
+    overrides. The first workspace is the workspace context where an eligible
+    account first asks for tour state; subsequent workspaces are opt-in only.
+    """
+
+    __tablename__ = "account_step_by_step_tour_states"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="account_step_by_step_tour_state_pkey"),
+        sa.UniqueConstraint("account_id", name="account_step_by_step_tour_state_account_id_key"),
+    )
+
+    id: Mapped[str] = mapped_column(
+        StringUUID,
+        insert_default=gen_uuidv7_string,
+        default_factory=gen_uuidv7_string,
+        init=False,
+    )
+    account_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    first_workspace_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True, default=None)
+    skipped: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"), default=False)
+    completed_task_ids: Mapped[list[str]] = mapped_column(AdjustedJSON, nullable=False, default_factory=list)
+    manually_enabled_workspace_ids: Mapped[list[str]] = mapped_column(
+        AdjustedJSON,
+        nullable=False,
+        default_factory=list,
+    )
+    manually_disabled_workspace_ids: Mapped[list[str]] = mapped_column(
+        AdjustedJSON,
+        nullable=False,
+        default_factory=list,
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        server_default=func.current_timestamp(),
+        nullable=False,
+        init=False,
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        server_default=func.current_timestamp(),
+        nullable=False,
+        init=False,
+        onupdate=func.current_timestamp(),
+    )

api/services/feature_service.py

@@ -182,6 +182,7 @@ class SystemFeatureModel(FeatureResponseModel):
     enable_trial_app: bool = False
     enable_explore_banner: bool = False
     enable_learn_app: bool = True
+    enable_step_by_step_tour: bool = False
     rbac_enabled: bool = False
 
 
@@ -284,6 +285,7 @@ class FeatureService:
         system_features.enable_trial_app = dify_config.ENABLE_TRIAL_APP
         system_features.enable_explore_banner = dify_config.ENABLE_EXPLORE_BANNER
         system_features.enable_learn_app = dify_config.ENABLE_LEARN_APP
+        system_features.enable_step_by_step_tour = dify_config.ENABLE_STEP_BY_STEP_TOUR
 
     @classmethod
     def _fulfill_trial_models_from_env(cls) -> list[str]:

api/services/step_by_step_tour_service.py

@@ -0,0 +1,225 @@
+"""Account-level Step-by-step Tour persistence."""
+
+from datetime import datetime
+from typing import NotRequired, TypedDict
+
+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
+
+STEP_BY_STEP_TOUR_TASK_IDS = frozenset(("home", "studio", "knowledge", "integration"))
+
+
+class StepByStepTourStateResponse(TypedDict):
+    eligible: bool
+    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 StepByStepTourPatch(TypedDict):
+    action: str
+    task_id: NotRequired[str | None]
+
+
+class StepByStepTourService:
+    """Coordinate persisted tour state with account eligibility rules."""
+
+    @classmethod
+    def get_state(
+        cls,
+        *,
+        account: Account,
+        current_tenant_id: str,
+        session: Session | scoped_session,
+    ) -> StepByStepTourStateResponse:
+        eligible = cls.is_eligible(account)
+        state = cls._get_state(account.id, session=session)
+
+        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)
+
+        return cls._build_response(eligible=eligible, state=state)
+
+    @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,
+        )
+
+        session.commit()
+        session.refresh(state)
+        return cls._build_response(eligible=cls.is_eligible(account), state=state)
+
+    @classmethod
+    def is_eligible(cls, account: Account) -> bool:
+        if not dify_config.ENABLE_STEP_BY_STEP_TOUR:
+            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
+
+    @classmethod
+    def _apply_action(
+        cls,
+        *,
+        state: AccountStepByStepTourState,
+        action: str,
+        task_id: str | None,
+        current_tenant_id: str,
+    ) -> None:
+        match action:
+            case "skip":
+                state.skipped = True
+                state.manually_enabled_workspace_ids = cls._remove_id(
+                    state.manually_enabled_workspace_ids,
+                    current_tenant_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)
+            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)
+            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,
+                )
+            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,
+                )
+            case _:
+                raise ValueError(f"Unsupported action: {action}")
+
+    @classmethod
+    def _build_response(
+        cls,
+        *,
+        eligible: bool,
+        state: AccountStepByStepTourState | None,
+    ) -> StepByStepTourStateResponse:
+        if state is None:
+            return {
+                "eligible": eligible,
+                "first_workspace_id": None,
+                "skipped": False,
+                "completed_task_ids": [],
+                "manually_enabled_workspace_ids": [],
+                "manually_disabled_workspace_ids": [],
+                "updated_at": None,
+            }
+
+        return {
+            "eligible": eligible,
+            "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": getattr(state, "updated_at", None),
+        }
+
+    @staticmethod
+    def _validate_task_id(task_id: str) -> None:
+        if task_id not in STEP_BY_STEP_TOUR_TASK_IDS:
+            raise ValueError(f"Unsupported task_id: {task_id}")
+
+    @classmethod
+    def _add_id(cls, values: list[str], value: str) -> list[str]:
+        normalized = cls._normalize_ids(values)
+        if value in normalized:
+            return normalized
+        return [*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]
+
+    @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

api/tests/unit_tests/controllers/console/test_onboarding.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from inspect import unwrap
+from unittest.mock import Mock, PropertyMock, patch
+
+import pytest
+from flask import Flask
+from pydantic import ValidationError
+
+from controllers.console import console_ns
+from controllers.console.onboarding import (
+    StepByStepTourStateApi,
+    StepByStepTourStatePatchPayload,
+)
+from extensions.ext_database import db
+from models.account import Account, AccountStatus
+from services.step_by_step_tour_service import StepByStepTourService
+
+
+def _account() -> Account:
+    account = Account(name="User", email="user@example.com", status=AccountStatus.ACTIVE)
+    account.id = "account-1"
+    return account
+
+
+def _state_response() -> dict[str, object]:
+    return {
+        "eligible": True,
+        "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 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)
+
+    api = StepByStepTourStateApi()
+    method = unwrap(api.get)
+
+    with app.test_request_context("/console/api/onboarding/step-by-step-tour/state", method="GET"):
+        result = method(api, "workspace-1", _account())
+
+    assert result == {
+        "eligible": True,
+        "first_workspace_id": "workspace-1",
+        "skipped": False,
+        "completed_task_ids": ["home"],
+        "manually_enabled_workspace_ids": [],
+        "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
+
+
+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)
+
+    api = StepByStepTourStateApi()
+    method = unwrap(api.patch)
+    payload = {"action": "complete_task", "task_id": "studio"}
+
+    with app.test_request_context(
+        "/console/api/onboarding/step-by-step-tour/state",
+        method="PATCH",
+        json=payload,
+    ):
+        with patch.object(type(console_ns), "payload", new_callable=PropertyMock, return_value=payload):
+            result = method(api, "workspace-1", _account())
+
+    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
+
+
+def test_patch_payload_rejects_non_action_fields() -> None:
+    with pytest.raises(ValidationError, match="Extra inputs are not permitted"):
+        StepByStepTourStatePatchPayload.model_validate({"action": "skip", "skipped": True})
+
+
+def test_patch_payload_rejects_task_id_without_task_action() -> None:
+    with pytest.raises(ValidationError, match="task_id is only supported for task actions"):
+        StepByStepTourStatePatchPayload.model_validate({"action": "skip", "task_id": "home"})
+
+
+def test_patch_payload_requires_action() -> None:
+    with pytest.raises(ValidationError):
+        StepByStepTourStatePatchPayload.model_validate({"task_id": "home"})

api/tests/unit_tests/services/test_feature_service_learn_app.py

@@ -6,6 +6,7 @@ from services.feature_service import FeatureService, SystemFeatureModel
 
 def test_system_feature_model_defaults_enable_learn_app():
     assert SystemFeatureModel().enable_learn_app is True
+    assert SystemFeatureModel().enable_step_by_step_tour is False
 
 
 @pytest.mark.parametrize("enabled", [True, False])
@@ -15,3 +16,14 @@ def test_get_system_features_reads_enable_learn_app(monkeypatch: pytest.MonkeyPa
     result = FeatureService.get_system_features()
 
     assert result.enable_learn_app is enabled
+
+
+@pytest.mark.parametrize("enabled", [True, False])
+def test_get_system_features_reads_enable_step_by_step_tour(
+    monkeypatch: pytest.MonkeyPatch, enabled: bool
+) -> None:
+    monkeypatch.setattr(feature_service_module.dify_config, "ENABLE_STEP_BY_STEP_TOUR", enabled)
+
+    result = FeatureService.get_system_features()
+
+    assert result.enable_step_by_step_tour is enabled

api/tests/unit_tests/services/test_step_by_step_tour_service.py

@@ -0,0 +1,242 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import pytest
+from sqlalchemy.exc import IntegrityError
+
+from models.account import Account, AccountStatus
+from models.onboarding import AccountStepByStepTourState
+from services import step_by_step_tour_service as service_module
+from services.step_by_step_tour_service import StepByStepTourService
+
+
+class _ScalarResult:
+    def __init__(self, state: AccountStepByStepTourState | None) -> None:
+        self._state = state
+
+    def scalar_one_or_none(self) -> AccountStepByStepTourState | None:
+        return self._state
+
+
+class _FakeSession:
+    def __init__(self, state: AccountStepByStepTourState | None = None) -> None:
+        self.state = state
+        self.added: list[AccountStepByStepTourState] = []
+        self.commit_count = 0
+        self.flush_count = 0
+        self.refresh_count = 0
+        self.rollback_count = 0
+
+    def execute(self, _stmt) -> _ScalarResult:
+        return _ScalarResult(self.state)
+
+    def add(self, state: AccountStepByStepTourState) -> None:
+        self.state = state
+        self.added.append(state)
+
+    def flush(self) -> None:
+        self.flush_count += 1
+
+    def commit(self) -> None:
+        self.commit_count += 1
+
+    def refresh(self, state: AccountStepByStepTourState) -> None:
+        self.refresh_count += 1
+        state.updated_at = datetime(2026, 6, 28, tzinfo=UTC)
+
+    def rollback(self) -> None:
+        self.rollback_count += 1
+
+
+class _RaceInsertSession(_FakeSession):
+    def __init__(self, state_after_rollback: AccountStepByStepTourState) -> None:
+        super().__init__(state=None)
+        self.state_after_rollback = state_after_rollback
+
+    def flush(self) -> None:
+        self.flush_count += 1
+        raise IntegrityError("insert", {}, Exception("duplicate"))
+
+    def rollback(self) -> None:
+        super().rollback()
+        self.state = self.state_after_rollback
+
+
+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 _set_tour_config(monkeypatch: pytest.MonkeyPatch, *, enabled: bool, rollout_started_at: datetime | None) -> None:
+    monkeypatch.setattr(service_module.dify_config, "ENABLE_STEP_BY_STEP_TOUR", enabled)
+    monkeypatch.setattr(service_module.dify_config, "STEP_BY_STEP_TOUR_ROLLOUT_STARTED_AT", rollout_started_at)
+
+
+def test_get_state_creates_state_and_records_first_workspace_for_eligible_account(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _set_tour_config(monkeypatch, enabled=True, rollout_started_at=datetime(2026, 6, 1))
+    session = _FakeSession()
+
+    result = StepByStepTourService.get_state(
+        account=_account(initialized_at=datetime(2026, 6, 28)),
+        current_tenant_id="workspace-1",
+        session=session,
+    )
+
+    assert result["eligible"] is True
+    assert result["first_workspace_id"] == "workspace-1"
+    assert result["completed_task_ids"] == []
+    assert len(session.added) == 1
+    assert session.added[0].account_id == "account-1"
+    assert session.commit_count == 1
+    assert session.refresh_count == 1
+
+
+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))
+    monkeypatch.setattr(service_module.dify_config, "EDITION", "SELF_HOSTED")
+
+    result = StepByStepTourService.is_eligible(_account(initialized_at=datetime(2026, 6, 28)))
+
+    assert result is True
+
+
+def test_get_state_does_not_create_state_for_ineligible_account_without_existing_state(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _set_tour_config(monkeypatch, enabled=True, rollout_started_at=datetime(2026, 6, 1))
+    session = _FakeSession()
+
+    result = StepByStepTourService.get_state(
+        account=_account(initialized_at=datetime(2026, 5, 31)),
+        current_tenant_id="workspace-1",
+        session=session,
+    )
+
+    assert result == {
+        "eligible": False,
+        "first_workspace_id": None,
+        "skipped": False,
+        "completed_task_ids": [],
+        "manually_enabled_workspace_ids": [],
+        "manually_disabled_workspace_ids": [],
+        "updated_at": None,
+    }
+    assert session.added == []
+    assert session.commit_count == 0
+
+
+def test_patch_state_persists_even_when_account_is_not_eligible(monkeypatch: pytest.MonkeyPatch) -> None:
+    _set_tour_config(monkeypatch, enabled=False, rollout_started_at=datetime(2026, 6, 1))
+    session = _FakeSession()
+
+    result = StepByStepTourService.patch_state(
+        account=_account(initialized_at=datetime(2026, 6, 28)),
+        current_tenant_id="workspace-2",
+        patch={"action": "enable_current_workspace"},
+        session=session,
+    )
+
+    assert result["eligible"] is False
+    assert result["skipped"] is False
+    assert result["manually_enabled_workspace_ids"] == ["workspace-2"]
+    assert result["manually_disabled_workspace_ids"] == []
+    assert len(session.added) == 1
+    assert session.commit_count == 1
+
+
+def test_patch_state_skip_action_sets_skipped_and_removes_current_workspace_enable(
+    monkeypatch: pytest.MonkeyPatch,
+) -> 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"]
+    session = _FakeSession(state=state)
+
+    result = StepByStepTourService.patch_state(
+        account=_account(initialized_at=datetime(2026, 6, 28)),
+        current_tenant_id="workspace-1",
+        patch={"action": "skip"},
+        session=session,
+    )
+
+    assert result["eligible"] is False
+    assert result["skipped"] is True
+    assert result["manually_enabled_workspace_ids"] == ["workspace-2"]
+    assert result["manually_disabled_workspace_ids"] == []
+    assert session.added == []
+    assert session.commit_count == 1
+
+
+def test_patch_state_disable_action_moves_current_workspace_to_disabled(
+    monkeypatch: pytest.MonkeyPatch,
+) -> 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"]
+    session = _FakeSession(state=state)
+
+    result = StepByStepTourService.patch_state(
+        account=_account(initialized_at=datetime(2026, 6, 28)),
+        current_tenant_id="workspace-1",
+        patch={"action": "disable_current_workspace"},
+        session=session,
+    )
+
+    assert result["manually_enabled_workspace_ids"] == ["workspace-2"]
+    assert result["manually_disabled_workspace_ids"] == ["workspace-1"]
+    assert session.commit_count == 1
+
+
+def test_patch_state_complete_and_uncomplete_task(monkeypatch: pytest.MonkeyPatch) -> None:
+    _set_tour_config(monkeypatch, enabled=False, rollout_started_at=datetime(2026, 6, 1))
+    state = _state()
+    state.completed_task_ids = ["home"]
+    session = _FakeSession(state=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=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=session,
+    )
+
+    assert result["completed_task_ids"] == ["studio"]
+
+
+def test_patch_state_recovers_when_concurrent_request_created_state(
+    monkeypatch: pytest.MonkeyPatch,
+) -> 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"]
+    session = _RaceInsertSession(state_after_rollback=existing_state)
+
+    result = StepByStepTourService.patch_state(
+        account=_account(initialized_at=datetime(2026, 6, 28)),
+        current_tenant_id="workspace-2",
+        patch={"action": "enable_current_workspace"},
+        session=session,
+    )
+
+    assert result["manually_enabled_workspace_ids"] == ["workspace-1", "workspace-2"]
+    assert session.flush_count == 1
+    assert session.rollback_count == 1
+    assert session.commit_count == 1