fix(workflow): align with GraphOn 0.7.0

Update the GraphOn PR pin and lockfile to its latest frame-aware runtime. Align worker scaling and container event typing with the new pause/resume scheduler while keeping Dify's Human Input adapters compatible.
chore: merge main into human input branch
2026-07-28 13:18:35 +08:00 · 2026-07-21 00:08:03 +08:00 · 2026-07-20 14:46:37 +00:00 · 2026-07-20 12:59:29 +00:00 · 2026-07-20 12:22:40 +00:00 · 2026-07-20 10:41:26 +00:00
275 changed files with 14316 additions and 2977 deletions
@@ -16,7 +16,7 @@ concurrency:
 jobs:
  api-unit:
    name: API Unit Tests
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    env:
      COVERAGE_FILE: coverage-unit
    defaults:
@@ -75,7 +75,7 @@ jobs:

  api-integration:
    name: API Integration Tests
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    env:
      COVERAGE_FILE: coverage-integration
      STORAGE_TYPE: opendal
@@ -129,7 +129,7 @@ jobs:

  api-coverage:
    name: API Coverage
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    needs:
      - api-unit
      - api-integration
@@ -173,7 +173,7 @@ jobs:

  create-manifest:
    needs: build
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    if: github.repository == 'langgenius/dify'
    strategy:
      matrix:
@@ -23,7 +23,7 @@ concurrency:
 jobs:
  validate:
    name: validate manifest + resolve target Dify release
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    if: github.repository == 'langgenius/dify'
    permissions:
      contents: read
@@ -87,7 +87,7 @@ jobs:
  release:
    name: build + attach standalone binaries (all targets)
    needs: validate
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    permissions:
      contents: write
    defaults:
@@ -9,7 +9,7 @@ concurrency:

 jobs:
  db-migration-test-postgres:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04

    steps:
      - name: Checkout code
@@ -59,7 +59,7 @@ jobs:
        run: uv run --directory api flask upgrade-db

  db-migration-test-mysql:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04

    steps:
      - name: Checkout code
@@ -13,7 +13,7 @@ on:

 jobs:
  deploy:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    if: |
      github.event.workflow_run.conclusion == 'success' &&
      github.event.workflow_run.head_branch == 'deploy/agent'
@@ -10,7 +10,7 @@ on:

 jobs:
  deploy:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    if: |
      github.event.workflow_run.conclusion == 'success' &&
      github.event.workflow_run.head_branch == 'deploy/dev'
@@ -13,7 +13,7 @@ on:

 jobs:
  deploy:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    if: |
      github.event.workflow_run.conclusion == 'success' &&
      github.event.workflow_run.head_branch == 'deploy/enterprise'
@@ -13,7 +13,7 @@ on:

 jobs:
  deploy:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    if: |
      github.event.workflow_run.conclusion == 'success' &&
      github.event.workflow_run.head_branch == 'deploy/saas'
@@ -22,7 +22,7 @@ concurrency:
 jobs:
  check-cherry-pick-provenance:
    name: Require cherry-pick provenance
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
        with:
@@ -7,7 +7,7 @@ jobs:
    permissions:
      contents: read
      pull-requests: write
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - uses: actions/labeler@b8dd2d9be0f68b860e7dae5dae7d772984eacd6d # v6.2.0
        with:
@@ -21,7 +21,7 @@ concurrency:
 jobs:
  pre_job:
    name: Skip Duplicate Checks
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    outputs:
      should_skip: ${{ steps.skip_check.outputs.should_skip || 'false' }}
    steps:
@@ -37,7 +37,7 @@ jobs:
    name: Check Changed Files
    needs: pre_job
    if: needs.pre_job.outputs.should_skip != 'true'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    outputs:
      api-changed: ${{ steps.changes.outputs.api }}
      cli-changed: ${{ steps.changes.outputs.cli }}
@@ -164,7 +164,7 @@ jobs:
      - pre_job
      - check-changes
    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.api-changed != 'true'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Report skipped API tests
        run: echo "No API-related changes detected; skipping API tests."
@@ -177,7 +177,7 @@ jobs:
      - check-changes
      - api-tests-run
      - api-tests-skip
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Finalize API Tests status
        env:
@@ -224,7 +224,7 @@ jobs:
      - pre_job
      - check-changes
    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.cli-changed != 'true'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Report skipped CLI tests
        run: echo "No CLI-related changes detected; skipping CLI tests."
@@ -237,7 +237,7 @@ jobs:
      - check-changes
      - cli-tests-run
      - cli-tests-skip
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Finalize CLI Tests status
        env:
@@ -284,7 +284,7 @@ jobs:
      - pre_job
      - check-changes
    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.web-changed != 'true'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Report skipped web tests
        run: echo "No web-related changes detected; skipping web tests."
@@ -297,7 +297,7 @@ jobs:
      - check-changes
      - web-tests-run
      - web-tests-skip
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Finalize Web Tests status
        env:
@@ -344,7 +344,7 @@ jobs:
      - pre_job
      - check-changes
    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.e2e-changed != 'true'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Report skipped web full-stack e2e
        run: echo "No E2E-related changes detected; skipping web full-stack E2E."
@@ -357,7 +357,7 @@ jobs:
      - check-changes
      - web-e2e-run
      - web-e2e-skip
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Finalize Web Full-Stack E2E status
        env:
@@ -411,7 +411,7 @@ jobs:
      - pre_job
      - check-changes
    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.vdb-changed != 'true'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Report skipped VDB tests
        run: echo "No VDB-related changes detected; skipping VDB tests."
@@ -424,7 +424,7 @@ jobs:
      - check-changes
      - vdb-tests-run
      - vdb-tests-skip
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Finalize VDB Tests status
        env:
@@ -470,7 +470,7 @@ jobs:
      - pre_job
      - check-changes
    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.migration-changed != 'true'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Report skipped DB migration tests
        run: echo "No migration-related changes detected; skipping DB migration tests."
@@ -483,7 +483,7 @@ jobs:
      - check-changes
      - db-migration-test-run
      - db-migration-test-skip
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Finalize DB Migration Test status
        env:
@@ -530,7 +530,7 @@ jobs:
      - pre_job
      - check-changes
    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.sandbox-runtime-changed != 'true'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Report skipped sandbox runtime tests
        run: echo "No sandbox-runtime-related changes detected; skipping sandbox runtime tests."
@@ -543,7 +543,7 @@ jobs:
      - check-changes
      - sandbox-runtime-tests-run
      - sandbox-runtime-tests-skip
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Finalize Sandbox Runtime Tests status
        env:
@@ -14,7 +14,7 @@ concurrency:
 jobs:
  check-changes:
    name: Check Changed Files
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    outputs:
      external-e2e-changed: ${{ steps.changes.outputs.external_e2e }}
    steps:
@@ -12,7 +12,7 @@ permissions: {}
 jobs:
  comment:
    name: Comment PR with pyrefly diff
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    permissions:
      actions: read
      contents: read
@@ -10,7 +10,7 @@ permissions:

 jobs:
  pyrefly-diff:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    permissions:
      contents: read
      issues: write
@@ -12,7 +12,7 @@ permissions: {}
 jobs:
  comment:
    name: Comment PR with type coverage
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    permissions:
      actions: read
      contents: read
@@ -10,7 +10,7 @@ permissions:

 jobs:
  pyrefly-type-coverage:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    permissions:
      contents: read
      issues: write
@@ -13,7 +13,7 @@ concurrency:
 jobs:
  sandbox-runtime-unit:
    name: Sandbox Runtime Unit Tests
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    defaults:
      run:
        shell: bash
@@ -37,7 +37,7 @@ jobs:

  sandbox-runtime-lint:
    name: Sandbox Runtime Lint
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    defaults:
      run:
        shell: bash
@@ -64,7 +64,7 @@ jobs:

  sandbox-runtime-integration:
    name: Sandbox Runtime Integration Tests
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    defaults:
      run:
        shell: bash
@@ -16,7 +16,7 @@ jobs:
    name: Validate PR title
    permissions:
      pull-requests: read
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Complete merge group check
        if: github.event_name == 'merge_group'
@@ -12,7 +12,7 @@ on:
 jobs:
  stale:

-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    permissions:
      issues: write
      pull-requests: write
@@ -19,7 +19,7 @@ permissions:
 jobs:
  python-style:
    name: Python Style
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04

    steps:
      - name: Checkout code
@@ -83,7 +83,7 @@ jobs:

  web-style:
    name: Web Style
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    defaults:
      run:
        working-directory: ./web
@@ -182,7 +182,7 @@ jobs:

  superlinter:
    name: SuperLinter
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04

    steps:
      - name: Checkout code
@@ -17,7 +17,7 @@ concurrency:
 jobs:
  build:
    name: unit test for Node.js SDK
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04

    defaults:
      run:
@@ -35,7 +35,7 @@ concurrency:
 jobs:
  translate:
    if: github.repository == 'langgenius/dify'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    timeout-minutes: 120

    steps:
@@ -16,7 +16,7 @@ concurrency:
 jobs:
  trigger:
    if: github.repository == 'langgenius/dify'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    timeout-minutes: 5

    steps:
@@ -16,7 +16,7 @@ jobs:
  test:
    name: Full VDB Tests
    if: github.repository == 'langgenius/dify'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    strategy:
      matrix:
        python-version:
@@ -13,7 +13,7 @@ concurrency:
 jobs:
  test:
    name: VDB Smoke Tests
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    strategy:
      matrix:
        python-version:
@@ -569,8 +569,8 @@ WORKFLOW_GENERATOR_NODE_BUILDER_MAX_WORKERS=6
 GRAPH_ENGINE_MIN_WORKERS=3
 # Maximum number of workers per GraphEngine instance (default: 10)
 GRAPH_ENGINE_MAX_WORKERS=10
-# Queue depth threshold that triggers worker scale up (default: 3)
-GRAPH_ENGINE_SCALE_UP_THRESHOLD=3
+# Pending task threshold that triggers worker scale up (default: 0)
+GRAPH_ENGINE_SCALE_UP_THRESHOLD=0
 # Seconds of idle time before scaling down workers (default: 5.0)
 GRAPH_ENGINE_SCALE_DOWN_IDLE_TIME=5.0

@@ -1,4 +1,4 @@
-from datetime import timedelta
+from datetime import datetime, timedelta
 from enum import StrEnum
 from typing import Literal

@@ -856,9 +856,9 @@ class WorkflowConfig(BaseSettings):
        default=10,
    )

-    GRAPH_ENGINE_SCALE_UP_THRESHOLD: PositiveInt = Field(
-        description="Queue depth threshold that triggers worker scale up",
-        default=3,
+    GRAPH_ENGINE_SCALE_UP_THRESHOLD: NonNegativeInt = Field(
+        description="Pending task threshold that triggers worker scale up",
+        default=0,
    )

    GRAPH_ENGINE_SCALE_DOWN_IDLE_TIME: float = Field(
@@ -1138,6 +1138,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):
    """
@@ -40,6 +40,7 @@ from . import (
    init_validate,
    knowledge_fs_proxy,
    notification,
+    onboarding,
    ping,
    setup,
    spec,
@@ -209,6 +210,7 @@ __all__ = [
    "notification",
    "oauth",
    "oauth_server",
+    "onboarding",
    "ops_trace",
    "parameter",
    "ping",
@@ -607,6 +607,13 @@ class DatasetListApi(Resource):
                    ReplaceMemberBindings(scope=RBACResourceWhitelistScope.ALL),
                )
                initialize_created_app_rbac_access_task.delay(current_tenant_id, current_user.id, dataset_id=dataset.id)
+            else:
+                enterprise_rbac_service.RBACService.DatasetAccess.replace_whitelist(
+                    current_tenant_id,
+                    current_user.id,
+                    dataset.id,
+                    ReplaceMemberBindings(scope=RBACResourceWhitelistScope.SPECIFIC),
+                )

        permission_keys_map = enterprise_rbac_service.RBACService.DatasetPermissions.batch_get(
            current_tenant_id,
@@ -875,7 +882,7 @@ class DatasetIndexingEstimateApi(Resource):
                file_details = session.scalars(
                    select(UploadFile).where(UploadFile.tenant_id == current_tenant_id, UploadFile.id.in_(file_ids))
                ).all()
-                if file_details is None:
+                if not file_details:
                    raise NotFound("File not found.")

                if file_details:
@@ -215,6 +215,7 @@ class ConsoleWorkflowEventsApi(Resource):
                    raise InvalidArgumentError(f"cannot subscribe to workflow run, workflow_run_id={workflow_run.id}")

            include_state_snapshot = request.args.get("include_state_snapshot", "false").lower() == "true"
+            continue_on_pause = request.args.get("continue_on_pause", "false").lower() == "true"

            def _generate_stream_events():
                if include_state_snapshot:
@@ -225,6 +226,8 @@ class ConsoleWorkflowEventsApi(Resource):
                            tenant_id=workflow_run.tenant_id,
                            app_id=workflow_run.app_id,
                            session_maker=session_maker,
+                            human_input_surface=HumanInputSurface.CONSOLE,
+                            close_on_pause=not continue_on_pause,
                        )
                    )
                return generator.convert_to_event_stream(
@@ -0,0 +1,106 @@
+"""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):
+    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,
+            ),
+        )
@@ -531,14 +531,22 @@ class DatasetListApi(DatasetApiResource):
        except services.errors.dataset.DatasetNameDuplicateError:
            raise DatasetNameDuplicateError()

-        if payload.permission == DatasetPermissionEnum.ALL_TEAM and dify_config.RBAC_ENABLED:
-            RBACService.DatasetAccess.replace_whitelist(
-                tenant_id,
-                current_user.id,
-                dataset.id,
-                ReplaceMemberBindings(scope=RBACResourceWhitelistScope.ALL),
-            )
-            initialize_created_app_rbac_access_task.delay(tenant_id, current_user.id, dataset_id=dataset.id)
+        if dify_config.RBAC_ENABLED:
+            if payload.permission == DatasetPermissionEnum.ALL_TEAM:
+                RBACService.DatasetAccess.replace_whitelist(
+                    tenant_id,
+                    current_user.id,
+                    dataset.id,
+                    ReplaceMemberBindings(scope=RBACResourceWhitelistScope.ALL),
+                )
+                initialize_created_app_rbac_access_task.delay(tenant_id, current_user.id, dataset_id=dataset.id)
+            else:
+                RBACService.DatasetAccess.replace_whitelist(
+                    tenant_id,
+                    current_user.id,
+                    dataset.id,
+                    ReplaceMemberBindings(scope=RBACResourceWhitelistScope.SPECIFIC),
+                )

        return _dump_service_dataset_detail(dataset, session=session), 200

@@ -86,6 +86,7 @@ class WorkflowEventsApi(WebApiResource):
                    raise InvalidArgumentError(f"cannot subscribe to workflow run, workflow_run_id={workflow_run.id}")

            include_state_snapshot = request.args.get("include_state_snapshot", "false").lower() == "true"
+            continue_on_pause = request.args.get("continue_on_pause", "false").lower() == "true"

            def _generate_stream_events():
                if include_state_snapshot:
@@ -96,6 +97,7 @@ class WorkflowEventsApi(WebApiResource):
                            tenant_id=app_model.tenant_id,
                            app_id=app_model.id,
                            session_maker=session_maker,
+                            close_on_pause=not continue_on_pause,
                        )
                    )
                return generator.convert_to_event_stream(
@@ -10,6 +10,7 @@ from core.app.entities.queue_entities import (
    QueueErrorEvent,
    QueueMessageEndEvent,
    QueueStopEvent,
+    QueueWorkflowPausedEvent,
 )
 from models.model import AppMode

@@ -43,7 +44,12 @@ class MessageBasedAppQueueManager(AppQueueManager):
        self._q.put(message)

        if isinstance(
-            event, QueueStopEvent | QueueErrorEvent | QueueMessageEndEvent | QueueAdvancedChatMessageEndEvent
+            event,
+            QueueStopEvent
+            | QueueErrorEvent
+            | QueueMessageEndEvent
+            | QueueAdvancedChatMessageEndEvent
+            | QueueWorkflowPausedEvent,
        ):
            self.stop_listen(execution_terminal=True)

@@ -9,6 +9,7 @@ from core.app.entities.queue_entities import (
    QueueStopEvent,
    QueueWorkflowFailedEvent,
    QueueWorkflowPartialSuccessEvent,
+    QueueWorkflowPausedEvent,
    QueueWorkflowSucceededEvent,
    WorkflowQueueMessage,
 )
@@ -39,6 +40,7 @@ class WorkflowAppQueueManager(AppQueueManager):
            | QueueMessageEndEvent
            | QueueWorkflowSucceededEvent
            | QueueWorkflowFailedEvent
+            | QueueWorkflowPausedEvent
            | QueueWorkflowPartialSuccessEvent,
        ):
            self.stop_listen(execution_terminal=True)
@@ -55,6 +55,7 @@ from core.workflow.variable_pool_initializer import add_variables_to_pool
 from core.workflow.workflow_entry import WorkflowEntry
 from core.workflow.workflow_run_outputs import project_node_outputs_for_workflow_run
 from graphon.entities.graph_config import NodeConfigDictAdapter
+from graphon.entities.pause_reason import HitlRequired
 from graphon.graph import Graph
 from graphon.graph_engine.layers import GraphEngineLayer
 from graphon.graph_events import (
@@ -433,7 +434,9 @@ class WorkflowBasedAppRunner:
                )
            case GraphRunPausedEvent():
                runtime_state = workflow_entry.graph_engine.graph_runtime_state
-                paused_nodes = runtime_state.get_paused_nodes()
+                paused_nodes = list(
+                    dict.fromkeys(reason.node_id for reason in event.reasons if isinstance(reason, HitlRequired))
+                )
                enriched_reasons = enrich_graph_pause_reasons(
                    reasons=event.reasons,
                    form_repository=HumanInputFormSubmissionRepository(),
@@ -23,7 +23,7 @@ from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeE
 from core.workflow.system_variables import SystemVariableKey
 from core.workflow.variable_prefixes import SYSTEM_VARIABLE_NODE_ID
 from core.workflow.workflow_run_outputs import project_node_outputs_for_workflow_run
-from graphon.entities import WorkflowExecution, WorkflowNodeExecution
+from graphon.entities import WorkflowExecution, WorkflowNodeExecution, WorkflowStartReason
 from graphon.enums import (
    WorkflowExecutionStatus,
    WorkflowNodeExecutionMetadataKey,
@@ -116,7 +116,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
    def on_event(self, event: GraphEngineEvent) -> None:
        match event:
            case GraphRunStartedEvent():
-                self._handle_graph_run_started()
+                self._handle_graph_run_started(event)
            case GraphRunSucceededEvent():
                self._handle_graph_run_succeeded(event)
            case GraphRunPartialSucceededEvent():
@@ -147,7 +147,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
    # ------------------------------------------------------------------
    # Graph-level handlers
    # ------------------------------------------------------------------
-    def _handle_graph_run_started(self) -> None:
+    def _handle_graph_run_started(self, event: GraphRunStartedEvent | None = None) -> None:
        execution_id = self._get_execution_id()
        workflow_execution = WorkflowExecution.new(
            id_=execution_id,
@@ -161,6 +161,10 @@ class WorkflowPersistenceLayer(GraphEngineLayer):

        self._workflow_execution_repository.save(workflow_execution)
        self._workflow_execution = workflow_execution
+        if event is not None and event.reason == WorkflowStartReason.RESUMPTION:
+            node_executions = self._workflow_node_execution_repository.get_by_workflow_execution(execution_id)
+            self._node_execution_cache = {execution.id: execution for execution in node_executions}
+            self._node_sequence = max((execution.index for execution in node_executions), default=0)

    def _handle_graph_run_succeeded(self, event: GraphRunSucceededEvent) -> None:
        execution = self._get_workflow_execution()
@@ -19,6 +19,7 @@ from graphon.model_runtime.entities import (
 )
 from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
 from graphon.runtime import VariablePool
+from graphon.variables.template_resolution import convert_template


 class AdvancedPromptTransform(PromptTransform):
@@ -171,7 +172,7 @@ class AdvancedPromptTransform(PromptTransform):
                            if k.startswith("#"):
                                vp.add(k[1:-1].split("."), v)
                        raw_prompt = raw_prompt.replace("{{#context#}}", context or "")
-                        prompt = vp.convert_template(raw_prompt).text
+                        prompt = convert_template(vp, raw_prompt).text
                    else:
                        parser = PromptTemplateParser(template=raw_prompt, with_variable_tmpl=self.with_variable_tmpl)
                        prompt_inputs: Mapping[str, str] = {k: inputs[k] for k in parser.variable_keys if k in inputs}
@@ -16,6 +16,7 @@ from core.repositories.factory import (
    OrderConfig,
    WorkflowNodeExecutionRepository,
 )
+from core.repositories.sqlalchemy_workflow_node_execution_repository import SQLAlchemyWorkflowNodeExecutionRepository
 from graphon.entities import WorkflowNodeExecution
 from models import Account, CreatorUserRole, EndUser
 from models.workflow import WorkflowNodeExecutionTriggeredFrom
@@ -36,7 +37,7 @@ class CeleryWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):

    Key features:
    - Asynchronous save operations using Celery tasks
-    - In-memory cache for immediate reads
+    - In-memory cache for immediate reads with database backfill across Celery tasks
    - Support for multi-tenancy through tenant/app filtering
    - Automatic retry and error handling through Celery
    """
@@ -49,6 +50,8 @@ class CeleryWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
    _creator_user_role: CreatorUserRole
    _execution_cache: dict[str, WorkflowNodeExecution]
    _workflow_execution_mapping: dict[str, list[str]]
+    _database_loaded_workflow_executions: set[str]
+    _sql_repository: SQLAlchemyWorkflowNodeExecutionRepository

    def __init__(
        self,
@@ -98,6 +101,14 @@ class CeleryWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):

        # Cache for mapping workflow_execution_ids to execution IDs for efficient retrieval
        self._workflow_execution_mapping = {}
+        self._database_loaded_workflow_executions = set()
+        self._sql_repository = SQLAlchemyWorkflowNodeExecutionRepository(
+            session_factory=self._session_factory,
+            tenant_id=tenant_id,
+            user=user,
+            app_id=app_id,
+            triggered_from=triggered_from,
+        )

        logger.info(
            "Initialized CeleryWorkflowNodeExecutionRepository for tenant %s, app %s, triggered_from %s",
@@ -156,7 +167,7 @@ class CeleryWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
        order_config: OrderConfig | None = None,
    ) -> Sequence[WorkflowNodeExecution]:
        """
-        Retrieve all workflow node executions for a workflow execution from cache.
+        Retrieve workflow node executions from cache after loading persisted history once.

        Args:
            workflow_execution_id: The workflow execution identifier
@@ -166,6 +177,25 @@ class CeleryWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
            A sequence of WorkflowNodeExecution instances
        """
        try:
+            if workflow_execution_id not in self._database_loaded_workflow_executions:
+                try:
+                    persisted_executions = self._sql_repository.get_by_workflow_execution(
+                        workflow_execution_id,
+                        order_config,
+                    )
+                except Exception:
+                    logger.exception(
+                        "Failed to load persisted workflow node executions for execution %s",
+                        workflow_execution_id,
+                    )
+                else:
+                    execution_ids = self._workflow_execution_mapping.setdefault(workflow_execution_id, [])
+                    for execution in persisted_executions:
+                        self._execution_cache.setdefault(execution.id, execution)
+                        if execution.id not in execution_ids:
+                            execution_ids.append(execution.id)
+                    self._database_loaded_workflow_executions.add(workflow_execution_id)
+
            # Get execution IDs for this workflow execution from cache
            execution_ids = self._workflow_execution_mapping.get(workflow_execution_id, [])

@@ -67,6 +67,7 @@ class FormCreateParams:
    # workflow_execution_id for chatflow runs; set alone (workflow_execution_id None)
    # for Agent v2 chat ask_human forms, which have no workflow run.
    conversation_id: str | None = None
+    form_id: str | None = None


 class HumanInputFormRecipientEntity(Protocol):
@@ -110,7 +111,7 @@ class HumanInputFormEntity(Protocol):


 class HumanInputFormRepository(Protocol):
-    def get_form(self, node_id: str) -> HumanInputFormEntity | None: ...
+    def get_form(self, node_id: str, *, form_id: str | None = None) -> HumanInputFormEntity | None: ...

    def create_form(self, params: FormCreateParams) -> HumanInputFormEntity: ...

@@ -460,8 +461,7 @@ class HumanInputFormRepositoryImpl:
            raise ValueError("a runtime human input form requires a workflow_execution_id or conversation_id")

        with session_factory.create_session() as session, session.begin():
-            # Generate unique form ID
-            form_id = str(uuidv7())
+            form_id = params.form_id or str(uuidv7())
            start_time = naive_utc_now()
            node_expiration = form_config.expiration_time(start_time)
            form_definition = FormDefinition(
@@ -546,7 +546,7 @@ class HumanInputFormRepositoryImpl:

        return _HumanInputFormEntityImpl(form_model=form_model, recipient_models=recipient_models)

-    def get_form(self, node_id: str) -> HumanInputFormEntity | None:
+    def get_form(self, node_id: str, *, form_id: str | None = None) -> HumanInputFormEntity | None:
        if self._workflow_execution_id is None:
            raise ValueError("workflow_execution_id is required to load runtime human input forms")

@@ -555,6 +555,8 @@ class HumanInputFormRepositoryImpl:
            HumanInputForm.node_id == node_id,
            HumanInputForm.tenant_id == self._tenant_id,
        )
+        if form_id is not None:
+            form_query = form_query.where(HumanInputForm.id == form_id)
        with session_factory.create_session() as session:
            form_model: HumanInputForm | None = session.scalars(form_query).first()
            if form_model is None:
@@ -55,6 +55,7 @@ from core.tools.workflow_as_tool.provider import WorkflowToolProviderController
 from core.tools.workflow_as_tool.tool import WorkflowTool
 from extensions.ext_database import db
 from graphon.runtime import VariablePool
+from graphon.variables.template_resolution import convert_template
 from models.provider_ids import ToolProviderID
 from models.tools import ApiToolProvider, BuiltinToolProvider, WorkflowToolProvider
 from services.tools.mcp_tools_manage_service import MCPToolManageService
@@ -1113,7 +1114,7 @@ class ToolManager:
                    elif tool_input.type == "constant":
                        parameter_value = tool_input.value
                    elif tool_input.type == "mixed":
-                        segment_group = variable_pool.convert_template(str(tool_input.value))
+                        segment_group = convert_template(variable_pool, str(tool_input.value))
                        parameter_value = segment_group.text
                    else:
                        raise ToolParameterError(f"Unknown tool input type '{tool_input.type}'")
@@ -21,6 +21,7 @@ from graphon.enums import BuiltinNodeTypes
 from graphon.nodes.base.variable_template_parser import VariableTemplateParser
 from graphon.runtime import VariablePool
 from graphon.variables.consts import SELECTORS_LENGTH
+from graphon.variables.template_resolution import convert_template


 class DeliveryMethodType(enum.StrEnum):
@@ -116,7 +117,7 @@ class EmailDeliveryConfig(BaseModel):
        templated_body = cls.replace_url_placeholder(body, url)
        if variable_pool is None:
            return templated_body
-        return variable_pool.convert_template(templated_body).text
+        return convert_template(variable_pool, templated_body).text

    @classmethod
    def render_markdown_body(cls, body: str) -> str:
@@ -361,6 +361,12 @@ class DifyNodeFactory(NodeFactory):
        self._agent_runtime_support = AgentRuntimeSupport()
        self._agent_message_transformer = AgentMessageTransformer()

+    def with_runtime_state(self, graph_runtime_state: "GraphRuntimeState") -> "DifyNodeFactory":
+        return DifyNodeFactory(
+            graph_init_params=self.graph_init_params,
+            graph_runtime_state=graph_runtime_state,
+        )
+
    @staticmethod
    def _resolve_dify_context(run_context: Mapping[str, Any]) -> DifyRunContext:
        raw_ctx = run_context.get(DIFY_RUN_CONTEXT_KEY)
@@ -394,6 +400,7 @@ class DifyNodeFactory(NodeFactory):
        # stay explicit and constructors receive the concrete typed payload.
        resolved_node_data = self._validate_resolved_node_data(node_class, node_data)
        node_type = node_data.type
+        node: Node | None = None
        node_init_kwargs_factories: Mapping[NodeType, Callable[[], dict[str, object]]] = {
            BuiltinNodeTypes.CODE: lambda: {
                "code_executor": self._code_executor,
@@ -412,7 +419,8 @@ class DifyNodeFactory(NodeFactory):
            },
            BuiltinNodeTypes.HUMAN_INPUT: lambda: {
                "hitl_callback": self._build_human_input_callback(
-                    node_data=DifyHumanInputNodeData.model_validate(adapted_node_config["data"])
+                    node_data=DifyHumanInputNodeData.model_validate(adapted_node_config["data"]),
+                    execution_id_getter=lambda: node.execution_id if node is not None else None,
                ),
            },
            BuiltinNodeTypes.LLM: lambda: self._build_llm_compatible_node_init_kwargs(
@@ -457,13 +465,14 @@ class DifyNodeFactory(NodeFactory):
        }
        node_init_kwargs = node_init_kwargs_factories.get(node_type, lambda: {})()
        constructor_node_data = resolved_node_data.model_dump(mode="python", by_alias=True)
-        return node_class(
+        node = node_class(
            node_id=node_id,
            data=constructor_node_data,
            graph_init_params=self.graph_init_params,
            graph_runtime_state=self.graph_runtime_state,
            **node_init_kwargs,
        )
+        return node

    @staticmethod
    def _validate_resolved_node_data(node_class: type[Node], node_data: BaseNodeData) -> BaseNodeData:
@@ -524,6 +533,7 @@ class DifyNodeFactory(NodeFactory):
        self,
        *,
        node_data: DifyHumanInputNodeData,
+        execution_id_getter: Callable[[], str | None],
    ) -> DifyHITLCallback:
        return DifyHITLCallback(
            form_repository=self._human_input_runtime.build_form_repository(),
@@ -532,6 +542,7 @@ class DifyNodeFactory(NodeFactory):
            delivery_methods=self._human_input_runtime._resolve_delivery_methods(node_data=node_data),
            display_in_ui=self._human_input_runtime._display_in_ui(node_data=node_data),
            file_reference_factory=self._file_reference_factory,
+            execution_id_getter=execution_id_getter,
        )

    def _build_llm_compatible_node_init_kwargs(
@@ -21,6 +21,7 @@ from core.workflow.system_variables import SystemVariableKey, get_system_text
 from extensions.ext_database import db
 from graphon.model_runtime.entities.model_entities import AIModelEntity, ModelType
 from graphon.runtime import VariablePool
+from graphon.variables.template_resolution import convert_template
 from models.model import Conversation

 from .entities import AgentNodeData, AgentOldVersionModelFeatures, ParamsAutoGenerated
@@ -67,7 +68,7 @@ class AgentRuntimeSupport:
                    except TypeError:
                        parameter_value = str(agent_input.value)

-                    segment_group = variable_pool.convert_template(parameter_value)
+                    segment_group = convert_template(variable_pool, parameter_value)
                    parameter_value = segment_group.log if for_log else segment_group.text
                    try:
                        if not isinstance(agent_input.value, str):
@@ -2,7 +2,7 @@ from __future__ import annotations

 import json
 import logging
-from collections.abc import Mapping, Sequence
+from collections.abc import Callable, Mapping, Sequence
 from datetime import datetime, timedelta
 from typing import Any

@@ -11,10 +11,10 @@ from core.repositories.human_input_repository import FormCreateParams, HumanInpu
 from core.workflow.human_input_adapter import DeliveryChannelConfig
 from core.workflow.node_runtime import DifyFileReferenceFactory
 from graphon.nodes.human_input.entities import Completed, Expired, HITLContext, HITLDecision, PauseRequested
-from graphon.runtime import VariablePool
 from graphon.runtime.graph_runtime_state_protocol import ReadOnlyVariablePool
 from graphon.variables.factory import build_segment
 from graphon.variables.segments import Segment
+from graphon.variables.template_resolution import convert_template
 from libs.datetime_utils import ensure_naive_utc, naive_utc_now

 from .entities import (
@@ -31,24 +31,13 @@ from .session_binding import default_session_binding
 logger = logging.getLogger(__name__)


-def _require_template_variable_pool(pool: ReadOnlyVariablePool) -> VariablePool:
-    """Return the concrete graphon pool required for template expansion."""
-    if isinstance(pool, VariablePool):
-        return pool
-
-    msg = "human input rendering requires graphon.runtime.VariablePool for template expansion"
-    raise TypeError(msg)
-
-
 def render_form_content_before_submission(
    node_data: HumanInputNodeData,
    *,
    variable_pool: ReadOnlyVariablePool,
 ) -> str:
    """Process form content by substituting runtime variables before pause."""
-    # NOTE(QuantumGhost): This is not ideal, we should expose
-    # VariablePool method in Graphon.
-    rendered_form_content = _require_template_variable_pool(variable_pool).convert_template(node_data.form_content)
+    rendered_form_content = convert_template(variable_pool, node_data.form_content)
    return rendered_form_content.markdown


@@ -91,6 +80,7 @@ class DifyHITLCallback:
        delivery_methods: Sequence[DeliveryChannelConfig] = (),
        display_in_ui: bool = False,
        file_reference_factory: DifyFileReferenceFactory | None = None,
+        execution_id_getter: Callable[[], str | None] | None = None,
    ) -> None:
        self._form_repository = form_repository
        self._session_binding = default_session_binding
@@ -100,11 +90,17 @@ class DifyHITLCallback:
        self._delivery_methods = tuple(delivery_methods)
        self._display_in_ui = display_in_ui
        self._file_reference_factory = file_reference_factory
+        self._execution_id_getter = execution_id_getter

    def __call__(self, ctx: HITLContext) -> HITLDecision:
-        form = self._form_repository.get_form(ctx.node_id)
+        form_id = self._execution_id_getter() if self._execution_id_getter is not None else None
+        form = (
+            self._form_repository.get_form(ctx.node_id, form_id=form_id)
+            if form_id is not None
+            else self._form_repository.get_form(ctx.node_id)
+        )
        if form is None:
-            created = self._create_form(ctx)
+            created = self._create_form(ctx, form_id=form_id)
            return PauseRequested(session_id=self._session_binding.issue_session_id_for_form(form_id=created.id))

        status = self._normalize_status(form.status)
@@ -163,7 +159,7 @@ class DifyHITLCallback:
            outputs=outputs,
        )

-    def _create_form(self, ctx: HITLContext) -> HumanInputFormEntity:
+    def _create_form(self, ctx: HITLContext, *, form_id: str | None = None) -> HumanInputFormEntity:
        params = FormCreateParams(
            workflow_execution_id=self._workflow_execution_id or ctx.workflow_execution_id,
            conversation_id=self._conversation_id,
@@ -181,6 +177,7 @@ class DifyHITLCallback:
                    variable_pool=ctx.variable_pool,
                )
            ),
+            form_id=form_id,
        )
        return self._form_repository.create_form(params)

@@ -25,7 +25,6 @@ from graphon.enums import (
 from graphon.model_runtime.entities.llm_entities import LLMUsage
 from graphon.model_runtime.utils.encoders import jsonable_encoder
 from graphon.node_events import NodeRunResult
-from graphon.nodes.base import LLMUsageTrackingMixin
 from graphon.nodes.base.node import Node
 from graphon.variables import (
    ArrayFileSegment,
@@ -33,6 +32,7 @@ from graphon.variables import (
    StringSegment,
 )
 from graphon.variables.segments import ArrayObjectSegment
+from graphon.variables.template_resolution import convert_template

 from .entities import (
    Condition,
@@ -64,7 +64,7 @@ def _normalize_metadata_filter_sequence_item(value: object) -> str:
    return value if isinstance(value, str) else str(value)


-class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeData]):
+class KnowledgeRetrievalNode(Node[KnowledgeRetrievalNodeData]):
    node_type = BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL

    # Instance attributes specific to LLMNode.
@@ -309,7 +309,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
            resolved_value: str | Sequence[str] | int | float | None
            match value:
                case str():
-                    segment_group = variable_pool.convert_template(value)
+                    segment_group = convert_template(variable_pool, value)
                    if len(segment_group.value) == 1:
                        resolved_value = _normalize_metadata_filter_scalar(segment_group.value[0].to_object())
                    else:
@@ -317,7 +317,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
                case _ if isinstance(value, Sequence) and all(isinstance(v, str) for v in value):
                    resolved_values: list[str] = []
                    for v in value:
-                        segment_group = variable_pool.convert_template(v)
+                        segment_group = convert_template(variable_pool, v)
                        if len(segment_group.value) == 1:
                            resolved_values.append(
                                _normalize_metadata_filter_sequence_item(segment_group.value[0].to_object())
@@ -2,6 +2,7 @@ import logging
 import time
 from collections.abc import Generator, Mapping, Sequence
 from typing import Any, TypedDict
+from uuid import uuid4

 from configs import dify_config
 from context import capture_current_context
@@ -26,7 +27,6 @@ from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add
 from core.workflow.variable_prefixes import ENVIRONMENT_VARIABLE_NODE_ID
 from extensions.otel.runtime import is_instrument_flag_enabled
 from factories import file_factory
-from graphon.entities import GraphInitParams
 from graphon.entities.graph_config import NodeConfigDictAdapter
 from graphon.errors import WorkflowNodeRunFailedError
 from graphon.file import File
@@ -38,7 +38,8 @@ from graphon.graph_engine.layers import DebugLoggingLayer, ExecutionLimitsLayer
 from graphon.graph_events import GraphEngineEvent, GraphNodeEventBase, GraphRunFailedEvent
 from graphon.nodes import BuiltinNodeTypes
 from graphon.nodes.base.node import Node
-from graphon.runtime import ChildGraphNotFoundError, GraphRuntimeState, VariablePool
+from graphon.nodes.container_effects import ContainerAwaitRequest
+from graphon.runtime import GraphRuntimeState, VariablePool
 from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader, load_into_variable_pool
 from models.workflow import Workflow

@@ -69,77 +70,6 @@ def iter_dify_graph_engine_events(
    )


-class _WorkflowChildEngineBuilder:
-    tenant_id: str
-
-    def __init__(self, *, tenant_id: str) -> None:
-        self.tenant_id = tenant_id
-
-    @staticmethod
-    def _has_node_id(graph_config: Mapping[str, Any], node_id: str) -> bool | None:
-        """
-        Return whether `graph_config["nodes"]` contains the given node id.
-
-        Returns `None` when the nodes payload shape is unexpected, so graph-level
-        validation can surface the original configuration error.
-        """
-        nodes = graph_config.get("nodes")
-        if not isinstance(nodes, list):
-            return None
-
-        for node in nodes:
-            if not isinstance(node, Mapping):
-                return None
-            current_id = node.get("id")
-            if isinstance(current_id, str) and current_id == node_id:
-                return True
-        return False
-
-    def build_child_engine(
-        self,
-        *,
-        workflow_id: str,
-        graph_init_params: GraphInitParams,
-        parent_graph_runtime_state: GraphRuntimeState,
-        root_node_id: str,
-        variable_pool: VariablePool | None = None,
-    ) -> GraphEngine:
-        """Build a child engine with a fresh runtime state and only child-safe layers."""
-        child_graph_runtime_state = GraphRuntimeState(
-            variable_pool=variable_pool if variable_pool is not None else parent_graph_runtime_state.variable_pool,
-            start_at=time.perf_counter(),
-            execution_context=parent_graph_runtime_state.execution_context,
-        )
-        node_factory = DifyNodeFactory(
-            graph_init_params=graph_init_params,
-            graph_runtime_state=child_graph_runtime_state,
-        )
-
-        graph_config = graph_init_params.graph_config
-        has_root_node = self._has_node_id(graph_config=graph_config, node_id=root_node_id)
-        if has_root_node is False:
-            raise ChildGraphNotFoundError(f"child graph root node '{root_node_id}' not found")
-
-        child_graph = Graph.init(
-            graph_config=graph_config,
-            node_factory=node_factory,
-            root_node_id=root_node_id,
-        )
-
-        command_channel = InMemoryChannel()
-        config = GraphEngineConfig()
-        child_engine = GraphEngine(
-            workflow_id=workflow_id,
-            graph=child_graph,
-            graph_runtime_state=child_graph_runtime_state,
-            command_channel=command_channel,
-            config=config,
-            child_engine_builder=self,
-        )
-        child_engine.layer(LLMQuotaLayer(tenant_id=self.tenant_id))
-        return child_engine
-
-
 class _NodeConfigDict(TypedDict):
    id: str
    width: int
@@ -208,8 +138,8 @@ class WorkflowEntry:
        self.command_channel = command_channel
        self._response_stream_filter = response_stream_filter or ResponseStreamFilter()
        execution_context = capture_current_context()
-        graph_runtime_state.execution_context = execution_context
-        self._child_engine_builder = _WorkflowChildEngineBuilder(tenant_id=tenant_id)
+        # ponytail: Graphon snapshots omit process-local context; use a public rebind API when Graphon exposes one.
+        graph_runtime_state._execution_context = execution_context
        self.graph_engine = GraphEngine(
            workflow_id=workflow_id,
            graph=graph,
@@ -221,7 +151,6 @@ class WorkflowEntry:
                scale_up_threshold=dify_config.GRAPH_ENGINE_SCALE_UP_THRESHOLD,
                scale_down_idle_time=dify_config.GRAPH_ENGINE_SCALE_DOWN_IDLE_TIME,
            ),
-            child_engine_builder=self._child_engine_builder,
        )

        # Add debug logging layer when in debug mode
@@ -271,7 +200,7 @@ class WorkflowEntry:
        user_inputs: Mapping[str, Any],
        variable_pool: VariablePool,
        variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
-    ) -> tuple[Node, Generator[GraphNodeEventBase, None, None]]:
+    ) -> tuple[Node, Generator[GraphNodeEventBase | ContainerAwaitRequest, None, None]]:
        """
        Single step run workflow node
        :param workflow: Workflow instance
@@ -419,7 +348,7 @@ class WorkflowEntry:
    @classmethod
    def run_free_node(
        cls, node_data: dict[str, Any], node_id: str, tenant_id: str, user_id: str, user_inputs: dict[str, Any]
-    ) -> tuple[Node, Generator[GraphNodeEventBase, None, None]]:
+    ) -> tuple[Node, Generator[GraphNodeEventBase | ContainerAwaitRequest, None, None]]:
        """
        Run free node

@@ -613,14 +542,14 @@ class WorkflowEntry:
                variable_pool.add([variable_node_id] + variable_key_list, input_value)

    @staticmethod
-    def _traced_node_run(node: Node) -> Generator[GraphNodeEventBase, None, None]:
+    def _traced_node_run(node: Node) -> Generator[GraphNodeEventBase | ContainerAwaitRequest, None, None]:
        """
        Wraps a node's run method with OpenTelemetry tracing and returns a generator.
        """
        # Wrap node.run() with ObservabilityLayer hooks to produce node-level spans
        layer = ObservabilityLayer()
        layer.on_graph_start()
-        node.ensure_execution_id()
+        node.bind_execution_id(str(uuid4()))

        def _gen():
            error: Exception | None = None
@@ -117,9 +117,12 @@ class RedisSubscriptionBase(Subscription):
                )
                continue

-            self._enqueue_message(payload_bytes)
            if payload_bytes == SIG_CLOSE:
-                break
+                # Close signals are broadcast to every subscriber on the topic.
+                # The closing subscription is already handled by the _closed check above.
+                continue
+
+            self._enqueue_message(payload_bytes)

        _logger.debug("%s listener thread stopped for channel %s", self._get_subscription_type().title(), self._topic)
        try:
@@ -128,11 +128,17 @@ class _StreamsSubscription(Subscription):
                                data_bytes = data.encode()
                            case bytes() | bytearray():
                                data_bytes = bytes(data)
-                        if data_bytes is not None:
-                            if data_bytes == SIG_CLOSE:
-                                break
-                            self._queue.put_nowait(data_bytes)
                        last_id = entry_id
+                        if data_bytes is None:
+                            continue
+                        if data_bytes == SIG_CLOSE:
+                            # Close signals share the stream with normal events. Ignore signals
+                            # emitted by another subscription while this one is still open.
+                            with self._lock:
+                                if self._closed:
+                                    break
+                            continue
+                        self._queue.put_nowait(data_bytes)
        finally:
            self._queue.put_nowait(self._SENTINEL)
            with self._lock:
@@ -0,0 +1,39 @@
+"""add step by step tour state
+
+Revision ID: b8c9d0e1f2a3
+Revises: 3c9f8e2a1d7b
+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 = "b8c9d0e1f2a3"
+down_revision = "3c9f8e2a1d7b"
+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")
@@ -101,6 +101,7 @@ from .model import (
    UploadFile,
 )
 from .oauth import DatasourceOauthParamConfig, DatasourceProvider, OAuthAccessToken
+from .onboarding import AccountStepByStepTourState
 from .provider import (
    LoadBalancingModelConfig,
    Provider,
@@ -155,6 +156,7 @@ __all__ = [
    "Account",
    "AccountIntegrate",
    "AccountStatus",
+    "AccountStepByStepTourState",
    "AccountTrialAppRecord",
    "Agent",
    "AgentConfigDraft",
@@ -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(),
+    )
@@ -7787,6 +7787,30 @@ Initiate OAuth login process
 | ---- | ----------- | ------ |
 | 200 | Success | **application/json**: [OAuthProviderTokenResponse](#oauthprovidertokenresponse)<br> |

+### [GET] /onboarding/step-by-step-tour/state
+Get account-level Step-by-step Tour state
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 200 | Success | **application/json**: [StepByStepTourStateResponse](#stepbysteptourstateresponse)<br> |
+
+### [PATCH] /onboarding/step-by-step-tour/state
+Update account-level Step-by-step Tour state
+
+#### Request Body
+
+| Required | Schema |
+| -------- | ------ |
+|  Yes | **application/json**: [StepByStepTourStatePatchPayload](#stepbysteptourstatepatchpayload)<br> |
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 200 | Success | **application/json**: [StepByStepTourStateResponse](#stepbysteptourstateresponse)<br> |
+
 ### [DELETE] /rag/pipeline/customized/templates/{template_id}
 #### Parameters

@@ -21788,6 +21812,24 @@ Query parameters for listing snippet published workflows.
 | paused | integer |  | Yes |
 | success | integer |  | Yes |

+#### StepByStepTourStatePatchPayload
+
+| 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 |
+
+#### StepByStepTourStateResponse
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| completed_task_ids | [ string, <br>**Available values:** "home", "integration", "knowledge", "studio" ] |  | No |
+| first_workspace_id | string |  | No |
+| manually_disabled_workspace_ids | [ string ] |  | No |
+| manually_enabled_workspace_ids | [ string ] |  | No |
+| skipped | boolean |  | No |
+| updated_at | string |  | No |
+
 #### Storage

 | Name | Type | Description | Required |
@@ -21941,6 +21983,7 @@ Model class for provider system configuration response.
 | enable_learn_app | boolean, <br>**Default:** true |  | Yes |
 | enable_marketplace | boolean |  | Yes |
 | enable_social_oauth_login | boolean |  | Yes |
+| enable_step_by_step_tour | boolean |  | Yes |
 | enable_trial_app | boolean |  | Yes |
 | is_allow_create_workspace | boolean |  | Yes |
 | is_allow_register | boolean |  | Yes |
@@ -1577,6 +1577,7 @@ Default configuration for form inputs.
 | enable_learn_app | boolean, <br>**Default:** true |  | Yes |
 | enable_marketplace | boolean |  | Yes |
 | enable_social_oauth_login | boolean |  | Yes |
+| enable_step_by_step_tour | boolean |  | Yes |
 | enable_trial_app | boolean |  | Yes |
 | is_allow_create_workspace | boolean |  | Yes |
 | is_allow_register | boolean |  | Yes |
@@ -45,7 +45,7 @@ dependencies = [
    "zstandard==0.25.0",
    # Emerging: newer and fast-moving, use compatible pins
    "fastopenapi[flask]==0.7.0",
-    "graphon==0.6.0",
+    "graphon==0.7.0",
    "httpx-sse==0.4.3",
    "json-repair==0.60.1",
 ]
@@ -67,6 +67,7 @@ exclude = ["providers/vdb/__pycache__", "providers/trace/__pycache__"]
 [tool.uv.sources]
 dify-agent = { path = "../dify-agent", editable = true }
 flask-restx = { git = "https://github.com/asukaminato0721/flask-restx", rev = "27758e26f8f740d7525d5039c51a9e524b6e2b68" }
+graphon = { git = "https://github.com/langgenius/graphon", rev = "d48c36fb02d8aa0d31dc6a9140a27c04a370600f" }
 dify-vdb-alibabacloud-mysql = { workspace = true }
 dify-vdb-analyticdb = { workspace = true }
 dify-vdb-baidu = { workspace = true }
@@ -183,6 +183,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


@@ -285,6 +286,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]:
@@ -49,6 +49,7 @@ from graphon.errors import WorkflowNodeRunFailedError
 from graphon.graph_events import GraphNodeEventBase, NodeRunFailedEvent, NodeRunSucceededEvent
 from graphon.node_events import NodeRunResult
 from graphon.nodes.base.node import Node
+from graphon.nodes.container_effects import ContainerAwaitRequest
 from graphon.nodes.http_request import HTTP_REQUEST_CONFIG_FILTER_KEY, build_http_request_config
 from graphon.runtime import VariablePool
 from graphon.variables.variables import Variable, VariableBase
@@ -909,7 +910,10 @@ class RagPipelineService:

    def _handle_node_run_result(
        self,
-        getter: Callable[[], tuple[Node, Generator[GraphNodeEventBase, None, None]]],
+        getter: Callable[
+            [],
+            tuple[Node, Generator[GraphNodeEventBase | ContainerAwaitRequest, None, None]],
+        ],
        start_at: float,
        tenant_id: str,
        node_id: str,
@@ -0,0 +1,221 @@
+"""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):
+    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(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(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,
+        *,
+        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,
+        }
+
+    @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
@@ -485,7 +485,6 @@ def _build_pause_event(
    variable_pool: ReadOnlyVariablePool | None = None
    if resumption_context is not None:
        state = GraphRuntimeState.from_snapshot(resumption_context.serialized_graph_runtime_state)
-        paused_nodes = state.get_paused_nodes()
        outputs = dict(WorkflowRuntimeTypeConverter().to_json_encodable(state.outputs or {}))
        variable_pool = state.variable_pool

@@ -493,6 +492,9 @@ def _build_pause_event(
        pause_entity.get_pause_reasons(),
        variable_pool=variable_pool,
    )
+    paused_nodes = list(
+        dict.fromkeys(reason.node_id for reason in resolved_pause_reasons if isinstance(reason, HumanInputRequired))
+    )
    reasons = [reason.model_dump(mode="json") for reason in resolved_pause_reasons]
    human_input_form_ids = [
        form_id
@@ -62,6 +62,7 @@ from graphon.graph_events import GraphNodeEventBase, NodeRunFailedEvent, NodeRun
 from graphon.node_events import NodeRunResult
 from graphon.nodes import BuiltinNodeTypes
 from graphon.nodes.base.node import Node
+from graphon.nodes.container_effects import ContainerAwaitRequest
 from graphon.nodes.http_request import HTTP_REQUEST_CONFIG_FILTER_KEY, build_http_request_config
 from graphon.nodes.start.entities import StartNodeData
 from graphon.runtime import VariablePool
@@ -1447,7 +1448,10 @@ class WorkflowService:

    def _handle_single_step_result(
        self,
-        invoke_node_fn: Callable[[], tuple[Node, Generator[GraphNodeEventBase, None, None]]],
+        invoke_node_fn: Callable[
+            [],
+            tuple[Node, Generator[GraphNodeEventBase | ContainerAwaitRequest, None, None]],
+        ],
        start_at: float,
        node_id: str,
    ) -> WorkflowNodeExecution:
@@ -1483,7 +1487,11 @@ class WorkflowService:
        return node_execution

    def _execute_node_safely(
-        self, invoke_node_fn: Callable[[], tuple[Node, Generator[GraphNodeEventBase, None, None]]]
+        self,
+        invoke_node_fn: Callable[
+            [],
+            tuple[Node, Generator[GraphNodeEventBase | ContainerAwaitRequest, None, None]],
+        ],
    ) -> tuple[Node, NodeRunResult | None, bool, str | None]:
        """
        Execute node safely and handle errors according to error strategy.
@@ -78,6 +78,7 @@ def test_dify_config(monkeypatch: pytest.MonkeyPatch):
    assert config.AGENT_SHELL_ENABLED is True
    assert config.SENTRY_TRACES_SAMPLE_RATE == 1.0
    assert config.TEMPLATE_TRANSFORM_MAX_LENGTH == 400_000
+    assert config.GRAPH_ENGINE_SCALE_UP_THRESHOLD == 0

    # annotated field with custom configured value
    assert config.HTTP_REQUEST_MAX_READ_TIMEOUT == 300
@@ -2,6 +2,7 @@ from __future__ import annotations

 from collections.abc import Iterator
 from inspect import unwrap
+from types import SimpleNamespace
 from unittest.mock import PropertyMock, patch

 import pytest
@@ -275,9 +276,9 @@ class TestCustomizedPipelineTemplateApi:
        assert (response, status) == ("", 204)
        assert deleted_templates == [("template-1", tenant_id)]

+    @pytest.mark.parametrize("sqlite_session", [(PipelineCustomizedTemplate,)], indirect=True)
    def test_post_exports_yaml_from_orm_template(
-        self,
-        database_app: Flask,
+        self, app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine, sqlite_session: Session
    ) -> None:
        api = CustomizedPipelineTemplateApi()
        method = unwrap(api.post)
@@ -293,26 +294,29 @@ class TestCustomizedPipelineTemplateApi:
            language="en-US",
            created_by="00000000-0000-0000-0000-000000000002",
        )
-        db.session.add(template)
-        db.session.commit()
+        template.id = "template-1"
+        sqlite_session.add(template)
+        sqlite_session.commit()
+        monkeypatch.setattr(module, "db", SimpleNamespace(engine=sqlite_engine))

-        with database_app.test_request_context("/rag/pipeline/customized/templates/template-1", method="POST"):
-            response, status = method(api, template.id)
+        with app.test_request_context("/rag/pipeline/customized/templates/template-1", method="POST"):
+            response, status = method(api, "template-1")

        assert status == 200
        assert response == {"data": "dsl: value"}

+    @pytest.mark.parametrize("sqlite_session", [(PipelineCustomizedTemplate,)], indirect=True)
    def test_post_raises_when_template_is_missing(
-        self,
-        database_app: Flask,
+        self, app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine, sqlite_session: Session
    ) -> None:
        api = CustomizedPipelineTemplateApi()
        method = unwrap(api.post)
-        with (
-            database_app.test_request_context("/rag/pipeline/customized/templates/missing", method="POST"),
-            pytest.raises(ValueError, match="Customized pipeline template not found"),
-        ):
-            method(api, "44444444-4444-4444-4444-444444444444")
+        assert sqlite_session.get(PipelineCustomizedTemplate, "missing") is None
+        monkeypatch.setattr(module, "db", SimpleNamespace(engine=sqlite_engine))
+
+        with app.test_request_context("/rag/pipeline/customized/templates/missing", method="POST"):
+            with pytest.raises(ValueError, match="Customized pipeline template not found"):
+                method(api, "missing")


 class TestPublishCustomizedPipelineTemplateApi:
@@ -4,7 +4,7 @@ import json
 from datetime import UTC, datetime
 from inspect import unwrap
 from types import SimpleNamespace
-from unittest.mock import Mock
+from unittest.mock import ANY, Mock

 import pytest
 from flask import Flask, Response
@@ -18,6 +18,7 @@ from controllers.console.human_input_form import (
    WorkflowResponseConverter,
    _jsonify_form_definition,
 )
+from core.workflow.human_input_policy import HumanInputSurface
 from models.account import AccountStatus
 from models.enums import CreatorUserRole
 from models.human_input import RecipientType
@@ -344,3 +345,62 @@ def test_workflow_events_finished(app: Flask, monkeypatch: pytest.MonkeyPatch) -

    assert response.mimetype == "text/event-stream"
    assert "data" in response.get_data(as_text=True)
+
+
+def test_workflow_events_snapshot_can_continue_across_pauses(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
+    workflow_run = SimpleNamespace(
+        id="run-1",
+        created_by_role=CreatorUserRole.ACCOUNT,
+        created_by="user-1",
+        tenant_id="t1",
+        app_id="app-1",
+        finished_at=None,
+    )
+    app_model = SimpleNamespace(mode=AppMode.WORKFLOW)
+
+    class _RepoStub:
+        def get_workflow_run_by_id_and_tenant_id(self, **_kwargs):
+            return workflow_run
+
+    workflow_generator = Mock()
+    workflow_generator.convert_to_event_stream.return_value = iter(["data: snapshot\n\n"])
+    snapshot_builder = Mock(return_value=["snapshot-events"])
+
+    monkeypatch.setattr(
+        DifyAPIRepositoryFactory,
+        "create_api_workflow_run_repository",
+        lambda *_args, **_kwargs: _RepoStub(),
+    )
+    monkeypatch.setattr(
+        "controllers.console.human_input_form._retrieve_app_for_workflow_run",
+        lambda *_args, **_kwargs: app_model,
+    )
+    monkeypatch.setattr(
+        "controllers.console.human_input_form.WorkflowAppGenerator",
+        lambda: workflow_generator,
+    )
+    monkeypatch.setattr(
+        "controllers.console.human_input_form.build_workflow_event_stream",
+        snapshot_builder,
+    )
+    monkeypatch.setattr("controllers.console.human_input_form.db", SimpleNamespace(engine=object()))
+
+    api = ConsoleWorkflowEventsApi()
+    handler = unwrap(api.get)
+
+    with app.test_request_context(
+        "/console/api/workflow/run-1/events?include_state_snapshot=true&continue_on_pause=true",
+        method="GET",
+    ):
+        response = handler(api, "t1", SimpleNamespace(id="user-1"), workflow_run_id="run-1")
+
+    assert response.get_data(as_text=True) == "data: snapshot\n\n"
+    snapshot_builder.assert_called_once_with(
+        app_mode=AppMode.WORKFLOW,
+        workflow_run=workflow_run,
+        tenant_id="t1",
+        app_id="app-1",
+        session_maker=ANY,
+        human_input_surface=HumanInputSurface.CONSOLE,
+        close_on_pause=False,
+    )
@@ -0,0 +1,99 @@
+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 {
+        "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 == {
+        "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"})
@@ -1,3 +1,4 @@
+from types import SimpleNamespace
 from typing import override
 from unittest.mock import MagicMock, patch

@@ -5,6 +6,8 @@ import pytest
 from flask import Flask, request
 from flask_login import LoginManager, UserMixin
 from pydantic import BaseModel
+from sqlalchemy import select
+from sqlalchemy.orm import Session
 from werkzeug.exceptions import HTTPException

 from controllers.common.wraps import _extract_resource_id
@@ -34,6 +37,7 @@ from controllers.console.wraps import (
 )
 from models import Account
 from models.account import AccountStatus, TenantAccountRole
+from models.dataset import RateLimitLog
 from services.feature_service import LicenseStatus


@@ -657,8 +661,7 @@ class TestRateLimiting:
    """Test rate limiting decorator"""

    @patch("controllers.console.wraps.redis_client")
-    @patch("controllers.console.wraps.db")
-    def test_should_allow_requests_within_rate_limit(self, mock_db: MagicMock, mock_redis: MagicMock):
+    def test_should_allow_requests_within_rate_limit(self, mock_redis: MagicMock):
        """Test that requests within rate limit are allowed"""
        # Arrange
        mock_rate_limit = MagicMock()
@@ -685,8 +688,13 @@ class TestRateLimiting:
        mock_redis.zremrangebyscore.assert_called_once()

    @patch("controllers.console.wraps.redis_client")
-    @patch("controllers.console.wraps.db")
-    def test_should_reject_requests_over_rate_limit(self, mock_db: MagicMock, mock_redis: MagicMock):
+    @pytest.mark.parametrize("sqlite_session", [(RateLimitLog,)], indirect=True)
+    def test_should_reject_requests_over_rate_limit(
+        self,
+        mock_redis: MagicMock,
+        sqlite_session: Session,
+        monkeypatch: pytest.MonkeyPatch,
+    ):
        """Test that requests over rate limit are rejected and logged"""
        # Arrange
        app = create_app_with_login()
@@ -696,8 +704,7 @@ class TestRateLimiting:
        mock_rate_limit.subscription_plan = "pro"
        mock_redis.zcard.return_value = 11  # Over limit

-        mock_session = MagicMock()
-        mock_db.session = mock_session
+        monkeypatch.setattr("controllers.console.wraps.db", SimpleNamespace(session=sqlite_session))

        @cloud_edition_billing_rate_limit_check("knowledge")
        def knowledge_request():
@@ -719,9 +726,11 @@ class TestRateLimiting:
                    assert exc_info.value.code == 403
                    assert "rate limit" in str(exc_info.value.description)

-                    # Verify rate limit log was created
-                    mock_session.add.assert_called_once()
-                    mock_session.commit.assert_called_once()
+                    rate_limit_log = sqlite_session.scalar(select(RateLimitLog))
+                    assert rate_limit_log is not None
+                    assert rate_limit_log.tenant_id == "tenant123"
+                    assert rate_limit_log.subscription_plan == "pro"
+                    assert rate_limit_log.operation == "knowledge"


 class TestCloudUtmRecord:
@@ -1,4 +1,5 @@
 import logging
+from collections.abc import Iterator
 from http import HTTPStatus
 from inspect import unwrap
 from io import BytesIO
@@ -6,6 +7,8 @@ from unittest.mock import ANY, MagicMock, patch

 import pytest
 from flask import Flask
+from sqlalchemy import Engine, event
+from sqlalchemy.orm import Session, scoped_session, sessionmaker
 from werkzeug.datastructures import FileStorage
 from werkzeug.exceptions import Unauthorized

@@ -36,6 +39,17 @@ from libs.datetime_utils import naive_utc_now
 from models.account import Account, Tenant, TenantCustomConfigDict, TenantStatus


+@pytest.fixture
+def workspace_session(sqlite_engine: Engine) -> Iterator[scoped_session[Session]]:
+    """Provide the callable scoped session expected by Flask-SQLAlchemy controllers."""
+    Tenant.metadata.create_all(sqlite_engine, tables=[Tenant.__table__])
+    session = scoped_session(sessionmaker(bind=sqlite_engine, expire_on_commit=False))
+    try:
+        yield session
+    finally:
+        session.remove()
+
+
 def make_account(account_id: str = "u1") -> Account:
    account = Account(name="Test User", email=f"{account_id}@example.com")
    account.id = account_id
@@ -370,11 +384,13 @@ class TestTenantInfoResponse:


 class TestSwitchWorkspaceApi:
-    def test_switch_success(self, app: Flask):
+    def test_switch_success(self, app: Flask, workspace_session: scoped_session[Session]):
        api = SwitchWorkspaceApi()
        method = unwrap(api.post)
        payload = {"tenant_id": "t2"}
        tenant = make_tenant("t2")
+        workspace_session.add(tenant)
+        workspace_session.commit()
        user = make_account()
        with (
            app.test_request_context("/workspaces/switch", json=payload),
@@ -383,11 +399,10 @@ class TestSwitchWorkspaceApi:
                "controllers.console.workspace.workspace.WorkspaceService.get_tenant_info", return_value={"id": "t2"}
            ),
        ):
-            session = MagicMock()
-            session.get.return_value = tenant
-            result = method(api, session, user)
+            result = method(api, workspace_session, user)
+
        assert result["result"] == "success"
-        switch_tenant.assert_called_once_with(user, "t2", session=session)
+        switch_tenant.assert_called_once_with(user, "t2", session=workspace_session)

    def test_switch_not_linked(self, app: Flask):
        api = SwitchWorkspaceApi()
@@ -401,7 +416,7 @@ class TestSwitchWorkspaceApi:
            with pytest.raises(AccountNotLinkTenantError):
                method(api, MagicMock(), user)

-    def test_switch_tenant_not_found(self, app: Flask):
+    def test_switch_tenant_not_found(self, app: Flask, workspace_session: scoped_session[Session]):
        api = SwitchWorkspaceApi()
        method = unwrap(api.post)
        payload = {"tenant_id": "missing"}
@@ -410,19 +425,21 @@ class TestSwitchWorkspaceApi:
            app.test_request_context("/workspaces/switch", json=payload),
            patch("controllers.console.workspace.workspace.TenantService.switch_tenant"),
        ):
-            session = MagicMock()
-            session.get.return_value = None
            with pytest.raises(ValueError):
-                method(api, session, user)
+                method(api, workspace_session, user)


 class TestCustomConfigWorkspaceApi:
-    def test_post_success(self, app: Flask):
+    def test_post_success(self, app: Flask, workspace_session: scoped_session[Session]):
        api = CustomConfigWorkspaceApi()
        method = unwrap(api.post)
        tenant = make_tenant(custom_config={})
+        workspace_session.add(tenant)
+        workspace_session.commit()
+
        payload = {"remove_webapp_brand": True}
        events = []
+        event.listen(workspace_session, "after_commit", lambda _: events.append("commit"))
        with (
            app.test_request_context("/workspaces/custom-config", json=payload),
            patch(
@@ -430,27 +447,29 @@ class TestCustomConfigWorkspaceApi:
                side_effect=lambda *args, **kwargs: events.append("get_tenant_info") or {"id": "t1"},
            ),
        ):
-            session = MagicMock()
-            session.get.return_value = tenant
-            session.commit.side_effect = lambda: events.append("commit")
-            result = method(api, session, "t1")
+            result = method(api, workspace_session, "t1")
        assert result["result"] == "success"
        assert events == ["commit", "get_tenant_info"]

-    def test_logo_fallback(self, app: Flask):
+    def test_logo_fallback(self, app: Flask, workspace_session: scoped_session[Session]):
        api = CustomConfigWorkspaceApi()
        method = unwrap(api.post)
+
        tenant = make_tenant(custom_config={"replace_webapp_logo": "old-logo"})
+        workspace_session.add(tenant)
+        workspace_session.commit()
+
        payload = {"remove_webapp_brand": False}
+
        with (
            app.test_request_context("/workspaces/custom-config", json=payload),
            patch(
-                "controllers.console.workspace.workspace.WorkspaceService.get_tenant_info", return_value={"id": "t1"}
+                "controllers.console.workspace.workspace.WorkspaceService.get_tenant_info",
+                return_value={"id": "t1"},
            ),
        ):
-            session = MagicMock()
-            session.get.return_value = tenant
-            result = method(api, session, "t1")
+            result = method(api, workspace_session, "t1")
+
        assert tenant.custom_config_dict["replace_webapp_logo"] == "old-logo"
        assert result["result"] == "success"

@@ -541,14 +560,19 @@ class TestWebappLogoWorkspaceApi:


 class TestWorkspaceInfoApi:
-    def test_post_success(self, app: Flask):
+    def test_post_success(self, app: Flask, workspace_session: scoped_session[Session]):
        api = WorkspaceInfoApi()
        method = unwrap(api.post)
        tenant = make_tenant()
+        workspace_session.add(tenant)
+        workspace_session.commit()
+
        payload = {"name": "New Name"}
        events = []
        with (
            app.test_request_context("/workspaces/info", json=payload),
+            patch("controllers.console.workspace.workspace.db.get_or_404", return_value=tenant),
+            patch("controllers.console.workspace.workspace.db.session", workspace_session),
            patch(
                "controllers.console.workspace.workspace.WorkspaceService.get_tenant_info",
                side_effect=lambda *args, **kwargs: (
@@ -245,8 +245,7 @@ def _build_resumption_context(task_id: str) -> WorkflowResumptionContext:
        workflow_execution_id="run-1",
    )
    runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=0.0)
-    runtime_state.register_paused_node("node-1")
-    runtime_state.outputs = {"result": "value"}
+    runtime_state.set_output("result", "value")
    wrapper = _WorkflowGenerateEntityWrapper(entity=generate_entity)
    return WorkflowResumptionContext(
        generate_entity=wrapper,
@@ -3,7 +3,7 @@
 from __future__ import annotations

 from types import SimpleNamespace
-from unittest.mock import MagicMock, patch
+from unittest.mock import ANY, MagicMock, Mock, patch

 import pytest
 from flask import Flask
@@ -11,6 +11,7 @@ from flask import Flask
 from controllers.common.errors import NotFoundError
 from controllers.web.workflow_events import WorkflowEventsApi
 from models.enums import CreatorUserRole
+from models.model import AppMode


 def _workflow_app() -> SimpleNamespace:
@@ -125,3 +126,39 @@ class TestWorkflowEventsApi:
            response = WorkflowEventsApi().get(_workflow_app(), _end_user(), "run-1")

        assert response.mimetype == "text/event-stream"
+
+    @patch("controllers.web.workflow_events.DifyAPIRepositoryFactory")
+    @patch("controllers.web.workflow_events.db")
+    def test_snapshot_stream_can_continue_across_pauses(
+        self, mock_db: MagicMock, mock_factory: MagicMock, app: Flask, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        mock_db.engine = "engine"
+        run = SimpleNamespace(
+            id="run-1",
+            app_id="app-1",
+            created_by_role=CreatorUserRole.END_USER,
+            created_by="eu-1",
+            finished_at=None,
+        )
+        mock_repo = MagicMock()
+        mock_repo.get_workflow_run_by_id_and_tenant_id.return_value = run
+        mock_factory.create_api_workflow_run_repository.return_value = mock_repo
+
+        workflow_generator = Mock()
+        workflow_generator.convert_to_event_stream.return_value = iter(["data: snapshot\n\n"])
+        snapshot_builder = Mock(return_value=["snapshot-events"])
+        monkeypatch.setattr("controllers.web.workflow_events.WorkflowAppGenerator", lambda: workflow_generator)
+        monkeypatch.setattr("controllers.web.workflow_events.build_workflow_event_stream", snapshot_builder)
+
+        with app.test_request_context("/workflow/run-1/events?include_state_snapshot=true&continue_on_pause=true"):
+            response = WorkflowEventsApi().get(_workflow_app(), _end_user(), "run-1")
+
+        assert response.get_data(as_text=True) == "data: snapshot\n\n"
+        snapshot_builder.assert_called_once_with(
+            app_mode=AppMode.WORKFLOW,
+            workflow_run=run,
+            tenant_id="tenant-1",
+            app_id="app-1",
+            session_maker=ANY,
+            close_on_pause=False,
+        )
@@ -1,8 +1,11 @@
 import json
+from collections.abc import Iterator
 from unittest.mock import MagicMock

 import pytest
 from pytest_mock import MockerFixture
+from sqlalchemy.engine import Engine
+from sqlalchemy.orm import Session

 from core.agent.cot_agent_runner import CotAgentRunner
 from core.agent.entities import AgentScratchpadUnit
@@ -26,7 +29,7 @@ class DummyRunner(CotAgentRunner):


@pytest.fixture
-def runner(mocker: MockerFixture):
+def runner(mocker: MockerFixture, sqlite_engine: Engine) -> Iterator[DummyRunner]:
    # Prevent BaseAgentRunner __init__ from hitting database
    mocker.patch(
        "core.agent.base_agent_runner.BaseAgentRunner.organize_agent_history",
@@ -81,9 +84,12 @@ def runner(mocker: MockerFixture):
    runner.agent_callback = None
    runner.memory = None
    runner.history_prompt_messages = []
-    runner.session = MagicMock()
+    runner.session = Session(sqlite_engine)

-    return runner
+    try:
+        yield runner
+    finally:
+        runner.session.close()


 class TestFillInputs:
@@ -1,9 +1,12 @@
 import json
+from collections.abc import Iterator
 from typing import Any
 from unittest.mock import MagicMock

 import pytest
 from pytest_mock import MockerFixture
+from sqlalchemy.engine import Engine
+from sqlalchemy.orm import Session

 from core.agent.errors import AgentMaxIterationError
 from core.agent.fc_agent_runner import FunctionCallAgentRunner
@@ -69,7 +72,7 @@ class DummyResult:


@pytest.fixture
-def runner(mocker: MockerFixture):
+def runner(mocker: MockerFixture, sqlite_engine: Engine) -> Iterator[FunctionCallAgentRunner]:
    # Completely bypass BaseAgentRunner __init__ to avoid DB / Flask context
    mocker.patch(
        "core.agent.base_agent_runner.BaseAgentRunner.__init__",
@@ -131,7 +134,7 @@ def runner(mocker: MockerFixture):
    runner._current_thoughts = []
    runner.files = []
    runner.agent_callback = MagicMock()
-    runner.session = MagicMock()
+    runner.session = Session(sqlite_engine)

    runner._init_prompt_tools = MagicMock(return_value=({}, []))
    runner.create_agent_thought = MagicMock(return_value="thought1")
@@ -139,7 +142,10 @@ def runner(mocker: MockerFixture):
    runner.recalc_llm_max_tokens = MagicMock()
    runner.update_prompt_message_tool = MagicMock()

-    return runner
+    try:
+        yield runner
+    finally:
+        runner.session.close()


 # ==============================
@@ -6,7 +6,12 @@ from core.app.apps.base_app_queue_manager import PublishFrom
 from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import InvokeFrom
-from core.app.entities.queue_entities import QueueErrorEvent, QueueMessageEndEvent, QueueStopEvent
+from core.app.entities.queue_entities import (
+    QueueErrorEvent,
+    QueueMessageEndEvent,
+    QueueStopEvent,
+    QueueWorkflowPausedEvent,
+)


 class TestMessageBasedAppQueueManager:
@@ -63,3 +68,21 @@ class TestMessageBasedAppQueueManager:
        manager._publish(QueueMessageEndEvent(), PublishFrom.TASK_PIPELINE)

        assert manager._q.qsize() == 1
+
+    def test_publish_pause_event_stops_listener_without_aborting_execution(self):
+        with patch("core.app.apps.base_app_queue_manager.redis_client") as mock_redis:
+            mock_redis.setex.return_value = True
+            manager = MessageBasedAppQueueManager(
+                task_id="t1",
+                user_id="u1",
+                invoke_from=InvokeFrom.DEBUGGER,
+                conversation_id="c1",
+                app_mode="advanced-chat",
+                message_id="m1",
+            )
+        manager.stop_listen = Mock()
+        manager._is_stopped = Mock(return_value=False)
+
+        manager._publish(QueueWorkflowPausedEvent(), PublishFrom.APPLICATION_MANAGER)
+
+        manager.stop_listen.assert_called_once_with(execution_terminal=True)
@@ -334,7 +334,6 @@ class TestWorkflowBasedAppRunner:
            variable_pool=VariablePool.from_bootstrap(system_variables=default_system_variables()),
            start_at=0.0,
        )
-        graph_runtime_state.register_paused_node("node-1")
        workflow_entry = SimpleNamespace(graph_engine=SimpleNamespace(graph_runtime_state=graph_runtime_state))

        emails: list[dict] = []
@@ -1,11 +1,16 @@
 from __future__ import annotations

-from unittest.mock import patch
+from unittest.mock import Mock, patch

 from core.app.apps.base_app_queue_manager import PublishFrom
 from core.app.apps.workflow.app_queue_manager import WorkflowAppQueueManager
 from core.app.entities.app_invoke_entities import InvokeFrom
-from core.app.entities.queue_entities import QueueMessageEndEvent, QueuePingEvent, QueueStopEvent
+from core.app.entities.queue_entities import (
+    QueueMessageEndEvent,
+    QueuePingEvent,
+    QueueStopEvent,
+    QueueWorkflowPausedEvent,
+)


 class TestWorkflowAppQueueManager:
@@ -36,6 +41,19 @@ class TestWorkflowAppQueueManager:

        manager._publish(QueuePingEvent(), PublishFrom.TASK_PIPELINE)

+    def test_publish_pause_event_stops_listener_without_aborting_execution(self):
+        manager = WorkflowAppQueueManager(
+            task_id="task",
+            user_id="user",
+            invoke_from=InvokeFrom.DEBUGGER,
+            app_mode="workflow",
+        )
+        manager.stop_listen = Mock()
+
+        manager._publish(QueueWorkflowPausedEvent(), PublishFrom.APPLICATION_MANAGER)
+
+        manager.stop_listen.assert_called_once_with(execution_terminal=True)
+
    def test_listener_close_aborts_unfinished_execution(self):
        with (
            patch("core.app.apps.base_app_queue_manager.redis_client") as redis_client,
@@ -9,7 +9,7 @@ from core.app.entities.app_invoke_entities import WorkflowAppGenerateEntity
 from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 from core.ops.ops_trace_manager import TraceTask, TraceTaskName
 from core.workflow.system_variables import SystemVariableKey, build_system_variables
-from graphon.entities import WorkflowNodeExecution
+from graphon.entities import WorkflowNodeExecution, WorkflowStartReason
 from graphon.entities.pause_reason import SchedulingPause
 from graphon.enums import (
    BuiltinNodeTypes,
@@ -32,6 +32,7 @@ from graphon.graph_events import (
    NodeRunStartedEvent,
    NodeRunSucceededEvent,
 )
+from graphon.model_runtime.entities.llm_entities import LLMUsage
 from graphon.node_events import NodeRunResult
 from graphon.runtime import GraphRuntimeState, ReadOnlyGraphRuntimeStateWrapper, VariablePool

@@ -40,6 +41,7 @@ class _RepoRecorder:
    def __init__(self) -> None:
        self.saved: list[object] = []
        self.saved_exec_data: list[object] = []
+        self.loaded: list[object] = []

    def save(self, entity):
        self.saved.append(entity)
@@ -47,6 +49,9 @@ class _RepoRecorder:
    def save_execution_data(self, entity):
        self.saved_exec_data.append(entity)

+    def get_by_workflow_execution(self, _workflow_execution_id):
+        return self.loaded
+

 def _naive_utc_now() -> datetime:
    return datetime.now(UTC).replace(tzinfo=None)
@@ -165,12 +170,45 @@ class TestWorkflowPersistenceLayer:

        assert exec_repo.saved

+    def test_resumption_restores_container_execution_before_terminal_event(self):
+        layer, _, node_repo, _ = _make_layer()
+        started_at = _naive_utc_now()
+        execution = WorkflowNodeExecution(
+            id="loop-exec",
+            workflow_id="workflow-id",
+            workflow_execution_id="run-id",
+            index=4,
+            node_id="loop",
+            node_type=BuiltinNodeTypes.LOOP,
+            title="Loop",
+            status=WorkflowNodeExecutionStatus.RUNNING,
+            created_at=started_at,
+        )
+        node_repo.loaded = [execution]
+
+        layer.on_event(GraphRunStartedEvent(reason=WorkflowStartReason.RESUMPTION))
+        layer.on_event(
+            NodeRunSucceededEvent(
+                id=execution.id,
+                node_id=execution.node_id,
+                node_type=execution.node_type,
+                start_at=started_at,
+                node_run_result=NodeRunResult(status=WorkflowNodeExecutionStatus.SUCCEEDED),
+            )
+        )
+
+        assert execution.status == WorkflowNodeExecutionStatus.SUCCEEDED
+        assert layer._next_node_sequence() == 5
+
    def test_handle_graph_run_succeeded_updates_execution(self):
        layer, exec_repo, _, runtime_state = _make_layer()
        layer._handle_graph_run_started()
-        runtime_state.total_tokens = 3
-        runtime_state.node_run_steps = 2
-        runtime_state.outputs = {"out": "v"}
+        usage = LLMUsage.empty_usage()
+        usage.total_tokens = 3
+        runtime_state.add_llm_usage(usage)
+        for _ in range(2):
+            runtime_state.increment_node_run_steps()
+        runtime_state.set_output("out", "v")

        layer._handle_graph_run_succeeded(GraphRunSucceededEvent(outputs={"ok": True}))

@@ -182,8 +220,11 @@ class TestWorkflowPersistenceLayer:
    def test_handle_graph_run_partial_succeeded_updates_execution(self):
        layer, exec_repo, _, runtime_state = _make_layer()
        layer._handle_graph_run_started()
-        runtime_state.total_tokens = 5
-        runtime_state.node_run_steps = 4
+        usage = LLMUsage.empty_usage()
+        usage.total_tokens = 5
+        runtime_state.add_llm_usage(usage)
+        for _ in range(4):
+            runtime_state.increment_node_run_steps()
        runtime_state._graph_execution = SimpleNamespace(exceptions_count=2)

        layer._handle_graph_run_partial_succeeded(
@@ -289,8 +330,11 @@ class TestWorkflowPersistenceLayer:
    def test_handle_graph_run_paused_updates_outputs(self):
        layer, exec_repo, _, runtime_state = _make_layer()
        layer._handle_graph_run_started()
-        runtime_state.total_tokens = 7
-        runtime_state.node_run_steps = 5
+        usage = LLMUsage.empty_usage()
+        usage.total_tokens = 7
+        runtime_state.add_llm_usage(usage)
+        for _ in range(5):
+            runtime_state.increment_node_run_steps()

        layer._handle_graph_run_paused(GraphRunPausedEvent(outputs={"pause": True}))

@@ -245,6 +245,60 @@ class TestCeleryWorkflowNodeExecutionRepository:
        # Should return empty list since nothing in cache
        assert len(result) == 0

+    def test_get_by_workflow_execution_loads_persisted_executions_on_cache_miss(
+        self, mock_session_factory, mock_account, sample_workflow_node_execution
+    ):
+        repo = CeleryWorkflowNodeExecutionRepository(
+            session_factory=mock_session_factory,
+            tenant_id=RESOURCE_TENANT_ID,
+            user=mock_account,
+            app_id="test-app",
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+        )
+        repo._sql_repository = Mock()
+        repo._sql_repository.get_by_workflow_execution.return_value = [sample_workflow_node_execution]
+
+        result = repo.get_by_workflow_execution(sample_workflow_node_execution.workflow_execution_id)
+
+        assert result == [sample_workflow_node_execution]
+        assert repo._execution_cache[sample_workflow_node_execution.id] is sample_workflow_node_execution
+        assert repo._workflow_execution_mapping[sample_workflow_node_execution.workflow_execution_id] == [
+            sample_workflow_node_execution.id
+        ]
+
+    @patch("core.repositories.celery_workflow_node_execution_repository.save_workflow_node_execution_task")
+    def test_get_by_workflow_execution_merges_database_and_newer_cache(
+        self, mock_task, mock_session_factory, mock_account, sample_workflow_node_execution
+    ):
+        repo = CeleryWorkflowNodeExecutionRepository(
+            session_factory=mock_session_factory,
+            tenant_id=RESOURCE_TENANT_ID,
+            user=mock_account,
+            app_id="test-app",
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+        )
+        persisted_current = sample_workflow_node_execution.model_copy(deep=True)
+        historical = sample_workflow_node_execution.model_copy(
+            update={
+                "id": str(uuid4()),
+                "node_execution_id": str(uuid4()),
+                "index": 0,
+                "node_id": "start",
+            }
+        )
+        sample_workflow_node_execution.status = WorkflowNodeExecutionStatus.SUCCEEDED
+        repo.save(sample_workflow_node_execution)
+        repo._sql_repository = Mock()
+        repo._sql_repository.get_by_workflow_execution.return_value = [persisted_current, historical]
+
+        result = repo.get_by_workflow_execution(
+            sample_workflow_node_execution.workflow_execution_id,
+            OrderConfig(order_by=["index"], order_direction="asc"),
+        )
+
+        assert [execution.id for execution in result] == [historical.id, sample_workflow_node_execution.id]
+        assert result[1] is sample_workflow_node_execution
+
    @patch("core.repositories.celery_workflow_node_execution_repository.save_workflow_node_execution_task")
    def test_cache_operations(self, mock_task, mock_session_factory, mock_account, sample_workflow_node_execution):
        """Test cache operations work correctly."""
@@ -3,7 +3,10 @@
 from __future__ import annotations

 from types import SimpleNamespace
-from unittest.mock import MagicMock, Mock, patch
+from unittest.mock import Mock, patch
+
+import pytest
+from sqlalchemy.orm import Session

 from core.app.app_config.entities import DatasetRetrieveConfigEntity
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -14,13 +17,14 @@ def _retrieve_config() -> DatasetRetrieveConfigEntity:
    return DatasetRetrieveConfigEntity(retrieve_strategy=DatasetRetrieveConfigEntity.RetrieveStrategy.MULTIPLE)


-def test_get_dataset_tools_returns_empty_for_empty_dataset_ids() -> None:
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_get_dataset_tools_returns_empty_for_empty_dataset_ids(sqlite_session: Session) -> None:
    # Arrange
    retrieve_config = _retrieve_config()

    # Act
    tools = DatasetRetrieverTool.get_dataset_tools(
-        session=MagicMock(),
+        session=sqlite_session,
        tenant_id="tenant",
        dataset_ids=[],
        retrieve_config=retrieve_config,
@@ -35,13 +39,14 @@ def test_get_dataset_tools_returns_empty_for_empty_dataset_ids() -> None:
    assert tools == []


-def test_get_dataset_tools_returns_empty_for_missing_retrieve_config() -> None:
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_get_dataset_tools_returns_empty_for_missing_retrieve_config(sqlite_session: Session) -> None:
    # Arrange
    dataset_ids = ["d1"]

    # Act
    tools = DatasetRetrieverTool.get_dataset_tools(
-        session=MagicMock(),
+        session=sqlite_session,
        tenant_id="tenant",
        dataset_ids=dataset_ids,
        retrieve_config=None,  # type: ignore[arg-type]
@@ -56,7 +61,8 @@ def test_get_dataset_tools_returns_empty_for_missing_retrieve_config() -> None:
    assert tools == []


-def test_get_dataset_tools_builds_tool_and_restores_strategy() -> None:
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_get_dataset_tools_builds_tool_and_restores_strategy(sqlite_session: Session) -> None:
    # Arrange
    retrieve_config = _retrieve_config()
    retrieval_tool = SimpleNamespace(name="dataset_tool", description="desc", run=lambda query: f"result:{query}")
@@ -66,7 +72,7 @@ def test_get_dataset_tools_builds_tool_and_restores_strategy() -> None:
    # Act
    with patch("core.tools.utils.dataset_retriever_tool.DatasetRetrieval", return_value=feature):
        tools = DatasetRetrieverTool.get_dataset_tools(
-            session=MagicMock(),
+            session=sqlite_session,
            tenant_id="tenant",
            dataset_ids=["d1"],
            retrieve_config=retrieve_config,
@@ -83,7 +89,7 @@ def test_get_dataset_tools_builds_tool_and_restores_strategy() -> None:
    assert retrieve_config.retrieve_strategy == DatasetRetrieveConfigEntity.RetrieveStrategy.MULTIPLE


-def _build_dataset_tool() -> tuple[DatasetRetrieverTool, SimpleNamespace]:
+def _build_dataset_tool(sqlite_session: Session) -> tuple[DatasetRetrieverTool, SimpleNamespace]:
    retrieval_tool = SimpleNamespace(
        name="dataset_tool",
        description="desc",
@@ -93,7 +99,7 @@ def _build_dataset_tool() -> tuple[DatasetRetrieverTool, SimpleNamespace]:
    feature.to_dataset_retriever_tool.return_value = [retrieval_tool]
    with patch("core.tools.utils.dataset_retriever_tool.DatasetRetrieval", return_value=feature):
        tools = DatasetRetrieverTool.get_dataset_tools(
-            session=MagicMock(),
+            session=sqlite_session,
            tenant_id="tenant",
            dataset_ids=["d1"],
            retrieve_config=_retrieve_config(),
@@ -106,9 +112,10 @@ def _build_dataset_tool() -> tuple[DatasetRetrieverTool, SimpleNamespace]:
    return tools[0], retrieval_tool


-def test_runtime_parameters_shape() -> None:
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_runtime_parameters_shape(sqlite_session: Session) -> None:
    # Arrange
-    tool, _ = _build_dataset_tool()
+    tool, _ = _build_dataset_tool(sqlite_session)

    # Act
    params = tool.get_runtime_parameters()
@@ -118,33 +125,36 @@ def test_runtime_parameters_shape() -> None:
    assert params[0].name == "query"


-def test_empty_query_behavior() -> None:
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_empty_query_behavior(sqlite_session: Session) -> None:
    # Arrange
-    tool, _ = _build_dataset_tool()
+    tool, _ = _build_dataset_tool(sqlite_session)

    # Act
-    empty_query = list(tool.invoke(session=MagicMock(), user_id="u", tool_parameters={}))
+    empty_query = list(tool.invoke(session=sqlite_session, user_id="u", tool_parameters={}))

    # Assert
    assert len(empty_query) == 1
    assert empty_query[0].message.text == "please input query"


-def test_query_invocation_result() -> None:
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_query_invocation_result(sqlite_session: Session) -> None:
    # Arrange
-    tool, _ = _build_dataset_tool()
+    tool, _ = _build_dataset_tool(sqlite_session)

    # Act
-    result = list(tool.invoke(session=MagicMock(), user_id="u", tool_parameters={"query": "hello"}))
+    result = list(tool.invoke(session=sqlite_session, user_id="u", tool_parameters={"query": "hello"}))

    # Assert
    assert len(result) == 1
    assert result[0].message.text == "result:hello"


-def test_validate_credentials() -> None:
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_validate_credentials(sqlite_session: Session) -> None:
    # Arrange
-    tool, _ = _build_dataset_tool()
+    tool, _ = _build_dataset_tool(sqlite_session)

    # Act
    result = tool.validate_credentials(credentials={}, parameters={}, format_only=False)
@@ -1,9 +1,12 @@
+"""MCP tool tests using real SQLite sessions for the ORM invocation contract."""
+
 from __future__ import annotations

 import base64
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch

 import pytest
+from sqlalchemy.orm import Session

 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.mcp.types import (
@@ -97,7 +100,8 @@ def test_mcp_tool_usage_extraction_helpers():
    assert derived.total_tokens == 0


-def test_mcp_tool_invoke_handles_content_types_and_structured_output():
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_mcp_tool_invoke_handles_content_types_and_structured_output(sqlite_session: Session):
    tool = _build_mcp_tool()
    img_data = base64.b64encode(b"img").decode()
    blob_data = base64.b64encode(b"blob").decode()
@@ -123,7 +127,7 @@ def test_mcp_tool_invoke_handles_content_types_and_structured_output():
    )

    with patch.object(MCPTool, "invoke_remote_mcp_tool", return_value=result):
-        messages = list(tool.invoke(session=MagicMock(), user_id="user-1", tool_parameters={"a": 1}))
+        messages = list(tool.invoke(session=sqlite_session, user_id="user-1", tool_parameters={"a": 1}))

    types = [m.type for m in messages]
    assert ToolInvokeMessage.MessageType.JSON in types
@@ -133,7 +137,8 @@ def test_mcp_tool_invoke_handles_content_types_and_structured_output():
    assert tool.latest_usage.total_tokens == 5


-def test_mcp_tool_invoke_raises_for_unsupported_embedded_resource():
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_mcp_tool_invoke_raises_for_unsupported_embedded_resource(sqlite_session: Session):
    tool = _build_mcp_tool()
    # Use model_construct to bypass pydantic validation and force unsupported resource path.
    bad_resource = EmbeddedResource.model_construct(type="resource", resource=object())
@@ -141,7 +146,7 @@ def test_mcp_tool_invoke_raises_for_unsupported_embedded_resource():

    with patch.object(MCPTool, "invoke_remote_mcp_tool", return_value=result):
        with pytest.raises(ToolInvokeError, match="Unsupported embedded resource type"):
-            list(tool.invoke(session=MagicMock(), user_id="user-1", tool_parameters={}))
+            list(tool.invoke(session=sqlite_session, user_id="user-1", tool_parameters={}))


 def test_mcp_tool_handle_none_parameter_filters_empty_values():
@@ -28,6 +28,7 @@ from graphon.variables.segments import (
    StringSegment,
    get_segment_discriminator,
 )
+from graphon.variables.template_resolution import convert_template
 from graphon.variables.types import SegmentType
 from graphon.variables.utils import (
    dumps_with_segments,
@@ -98,7 +99,7 @@ def test_segment_group_to_text():
    template = (
        "Hello, {{#sys.user_id#}}! Your query is {{#node_id.custom_query#}}. And your key is {{#env.secret_key#}}."
    )
-    segments_group = variable_pool.convert_template(template)
+    segments_group = convert_template(variable_pool, template)

    assert segments_group.text == "Hello, fake-user-id! Your query is fake-user-query. And your key is fake-secret-key."
    assert segments_group.log == (
@@ -112,7 +113,7 @@ def test_convert_constant_to_segment_group():
        system_variables=build_system_variables(user_id="1", app_id="1", workflow_id="1"),
    )
    template = "Hello, world!"
-    segments_group = variable_pool.convert_template(template)
+    segments_group = convert_template(variable_pool, template)
    assert segments_group.text == "Hello, world!"
    assert segments_group.log == "Hello, world!"

@@ -120,7 +121,7 @@ def test_convert_constant_to_segment_group():
 def test_convert_variable_to_segment_group():
    variable_pool = _build_variable_pool(system_variables=build_system_variables(user_id="fake-user-id"))
    template = "{{#sys.user_id#}}"
-    segments_group = variable_pool.convert_template(template)
+    segments_group = convert_template(variable_pool, template)
    assert segments_group.text == "fake-user-id"
    assert segments_group.log == "fake-user-id"
    assert isinstance(segments_group.value[0], StringVariable)
@@ -5,7 +5,7 @@ The factory follows the same config adaptation path as production
 implementations before instantiation.
 """

-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, override

 from core.workflow.human_input_adapter import adapt_node_config_for_graph
 from core.workflow.node_factory import DifyNodeFactory
@@ -76,6 +76,14 @@ class MockNodeFactory(DifyNodeFactory):
            BuiltinNodeTypes.CODE: MockCodeNode,
        }

+    @override
+    def with_runtime_state(self, graph_runtime_state: "GraphRuntimeState") -> "MockNodeFactory":
+        return MockNodeFactory(
+            graph_init_params=self.graph_init_params,
+            graph_runtime_state=graph_runtime_state,
+            mock_config=self.mock_config,
+        )
+
    def create_node(self, node_config: dict[str, Any] | NodeConfigDict) -> Node:
        """
        Create a node instance, using mock implementations for third-party service nodes.
@@ -615,69 +615,6 @@ class MockIterationNode(MockNodeMixin, IterationNode):
        """Return the version of this mock node."""
        return "1"

-    def _create_graph_engine(self, index: int, item: Any):
-        """Create a graph engine with MockNodeFactory instead of DifyNodeFactory."""
-        # Import dependencies
-        from graphon.entities import GraphInitParams
-        from graphon.graph import Graph
-        from graphon.graph_engine import GraphEngine, GraphEngineConfig
-        from graphon.graph_engine.command_channels import InMemoryChannel
-        from graphon.runtime import GraphRuntimeState
-
-        # Import our MockNodeFactory instead of DifyNodeFactory
-        from .test_mock_factory import MockNodeFactory
-
-        # Create GraphInitParams from node attributes
-        graph_init_params = GraphInitParams(
-            workflow_id=self.workflow_id,
-            graph_config=self.graph_config,
-            run_context=self.run_context,
-            call_depth=self.workflow_call_depth,
-        )
-
-        # Create a deep copy of the variable pool for each iteration
-        variable_pool_copy = self.graph_runtime_state.variable_pool.model_copy(deep=True)
-
-        # append iteration variable (item, index) to variable pool
-        variable_pool_copy.add([self._node_id, "index"], index)
-        variable_pool_copy.add([self._node_id, "item"], item)
-
-        # Create a new GraphRuntimeState for this iteration
-        graph_runtime_state_copy = GraphRuntimeState(
-            variable_pool=variable_pool_copy,
-            start_at=self.graph_runtime_state.start_at,
-            total_tokens=0,
-            node_run_steps=0,
-        )
-
-        # Create a MockNodeFactory with the same mock_config
-        node_factory = MockNodeFactory(
-            graph_init_params=graph_init_params,
-            graph_runtime_state=graph_runtime_state_copy,
-            mock_config=self.mock_config,  # Pass the mock configuration
-        )
-
-        # Initialize the iteration graph with the mock node factory
-        iteration_graph = Graph.init(
-            graph_config=self.graph_config, node_factory=node_factory, root_node_id=self._node_data.start_node_id
-        )
-
-        if not iteration_graph:
-            from graphon.nodes.iteration.exc import IterationGraphNotFoundError
-
-            raise IterationGraphNotFoundError("iteration graph not found")
-
-        # Create a new GraphEngine for this iteration
-        graph_engine = GraphEngine(
-            workflow_id=self.workflow_id,
-            graph=iteration_graph,
-            graph_runtime_state=graph_runtime_state_copy,
-            command_channel=InMemoryChannel(),  # Use InMemoryChannel for sub-graphs
-            config=GraphEngineConfig(),
-        )
-
-        return graph_engine
-

 class MockLoopNode(MockNodeMixin, LoopNode):
    """Mock implementation of LoopNode that preserves mock configuration."""
@@ -687,56 +624,6 @@ class MockLoopNode(MockNodeMixin, LoopNode):
        """Return the version of this mock node."""
        return "1"

-    def _create_graph_engine(self, start_at, root_node_id: str):
-        """Create a graph engine with MockNodeFactory instead of DifyNodeFactory."""
-        # Import dependencies
-        from graphon.entities import GraphInitParams
-        from graphon.graph import Graph
-        from graphon.graph_engine import GraphEngine, GraphEngineConfig
-        from graphon.graph_engine.command_channels import InMemoryChannel
-        from graphon.runtime import GraphRuntimeState
-
-        # Import our MockNodeFactory instead of DifyNodeFactory
-        from .test_mock_factory import MockNodeFactory
-
-        # Create GraphInitParams from node attributes
-        graph_init_params = GraphInitParams(
-            workflow_id=self.workflow_id,
-            graph_config=self.graph_config,
-            run_context=self.run_context,
-            call_depth=self.workflow_call_depth,
-        )
-
-        # Create a new GraphRuntimeState for this iteration
-        graph_runtime_state_copy = GraphRuntimeState(
-            variable_pool=self.graph_runtime_state.variable_pool,
-            start_at=start_at.timestamp(),
-        )
-
-        # Create a MockNodeFactory with the same mock_config
-        node_factory = MockNodeFactory(
-            graph_init_params=graph_init_params,
-            graph_runtime_state=graph_runtime_state_copy,
-            mock_config=self.mock_config,  # Pass the mock configuration
-        )
-
-        # Initialize the loop graph with the mock node factory
-        loop_graph = Graph.init(graph_config=self.graph_config, node_factory=node_factory, root_node_id=root_node_id)
-
-        if not loop_graph:
-            raise ValueError("loop graph not found")
-
-        # Create a new GraphEngine for this iteration
-        graph_engine = GraphEngine(
-            workflow_id=self.workflow_id,
-            graph=loop_graph,
-            graph_runtime_state=graph_runtime_state_copy,
-            command_channel=InMemoryChannel(),  # Use InMemoryChannel for sub-graphs
-            config=GraphEngineConfig(),
-        )
-
-        return graph_engine
-

 class MockTemplateTransformNode(MockNodeMixin, TemplateTransformNode):
    """Mock implementation of TemplateTransformNode for testing."""
@@ -51,53 +51,6 @@ from .test_mock_factory import MockNodeFactory
 logger = logging.getLogger(__name__)


-class _TableTestChildEngineBuilder:
-    def __init__(self, *, use_mock_factory: bool, mock_config: MockConfig | None) -> None:
-        self._use_mock_factory = use_mock_factory
-        self._mock_config = mock_config
-
-    def build_child_engine(
-        self,
-        *,
-        workflow_id: str,
-        graph_init_params: GraphInitParams,
-        parent_graph_runtime_state: GraphRuntimeState,
-        root_node_id: str,
-        variable_pool: VariablePool | None = None,
-    ) -> GraphEngine:
-        child_graph_runtime_state = GraphRuntimeState(
-            variable_pool=variable_pool if variable_pool is not None else parent_graph_runtime_state.variable_pool,
-            start_at=time.perf_counter(),
-            execution_context=parent_graph_runtime_state.execution_context,
-        )
-        if self._use_mock_factory:
-            node_factory = MockNodeFactory(
-                graph_init_params=graph_init_params,
-                graph_runtime_state=child_graph_runtime_state,
-                mock_config=self._mock_config,
-            )
-        else:
-            node_factory = DifyNodeFactory(
-                graph_init_params=graph_init_params,
-                graph_runtime_state=child_graph_runtime_state,
-            )
-
-        graph_config = graph_init_params.graph_config
-        child_graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=root_node_id)
-        if not child_graph:
-            raise ValueError("child graph not found")
-
-        child_engine = GraphEngine(
-            workflow_id=workflow_id,
-            graph=child_graph,
-            graph_runtime_state=child_graph_runtime_state,
-            command_channel=InMemoryChannel(),
-            config=GraphEngineConfig(),
-            child_engine_builder=self,
-        )
-        return child_engine
-
-
@dataclass
 class WorkflowTestCase:
    """Represents a single test case for table-driven testing."""
@@ -379,10 +332,6 @@ class TableTestRunner:
                    scale_up_threshold=self.graph_engine_scale_up_threshold,
                    scale_down_idle_time=self.graph_engine_scale_down_idle_time,
                ),
-                child_engine_builder=_TableTestChildEngineBuilder(
-                    use_mock_factory=test_case.use_auto_mock,
-                    mock_config=test_case.mock_config,
-                ),
            )

            # Execute and collect events
@@ -73,13 +73,15 @@ def _create_human_input_node(
        node_data=node_data,
        file_reference_factory=_TestFileReferenceFactory(),
    )
-    return HumanInputNode(
+    node = HumanInputNode(
        node_id=config["id"],
        data=node_data,
        graph_init_params=graph_init_params,
        graph_runtime_state=graph_runtime_state,
        hitl_callback=callback,
    )
+    node.bind_execution_id("00000000-0000-4000-8000-000000000001")
+    return node


 def _build_node(
@@ -1,95 +0,0 @@
-from collections.abc import Mapping
-from typing import Any
-
-import pytest
-
-from core.workflow.system_variables import default_system_variables
-from graphon.entities import GraphInitParams
-from graphon.nodes.iteration.entities import IterationNodeData
-from graphon.nodes.iteration.exc import IterationGraphNotFoundError
-from graphon.nodes.iteration.iteration_node import IterationNode
-from graphon.runtime import (
-    ChildEngineBuilderNotConfiguredError,
-    ChildGraphNotFoundError,
-    GraphRuntimeState,
-    VariablePool,
-)
-from tests.workflow_test_utils import build_test_graph_init_params
-
-
-class _MissingGraphBuilder:
-    def build_child_engine(
-        self,
-        *,
-        workflow_id: str,
-        graph_init_params: GraphInitParams,
-        parent_graph_runtime_state: GraphRuntimeState,
-        root_node_id: str,
-        variable_pool: VariablePool | None = None,
-    ) -> object:
-        raise ChildGraphNotFoundError(f"child graph root node '{root_node_id}' not found")
-
-
-def _build_runtime_state() -> GraphRuntimeState:
-    return GraphRuntimeState(
-        variable_pool=VariablePool.from_bootstrap(system_variables=default_system_variables(), user_inputs={}),
-        start_at=0.0,
-    )
-
-
-def _build_iteration_node(
-    *,
-    graph_config: Mapping[str, Any],
-    runtime_state: GraphRuntimeState,
-    start_node_id: str,
-) -> IterationNode:
-    init_params = build_test_graph_init_params(graph_config=graph_config)
-    return IterationNode(
-        node_id="iteration-node",
-        data=IterationNodeData(
-            type="iteration",
-            title="Iteration",
-            iterator_selector=["start", "items"],
-            output_selector=["iteration-node", "output"],
-            start_node_id=start_node_id,
-        ),
-        graph_init_params=init_params,
-        graph_runtime_state=runtime_state,
-    )
-
-
-def test_graph_runtime_state_raises_specific_error_when_child_builder_is_missing():
-    runtime_state = _build_runtime_state()
-    graph_init_params = build_test_graph_init_params()
-
-    with pytest.raises(ChildEngineBuilderNotConfiguredError):
-        runtime_state.create_child_engine(
-            workflow_id="workflow",
-            graph_init_params=graph_init_params,
-            root_node_id="root",
-        )
-
-
-def test_iteration_node_only_translates_child_graph_not_found_error():
-    runtime_state = _build_runtime_state()
-    runtime_state.bind_child_engine_builder(_MissingGraphBuilder())
-    node = _build_iteration_node(
-        graph_config={"nodes": [{"id": "present-node"}], "edges": []},
-        runtime_state=runtime_state,
-        start_node_id="missing-node",
-    )
-
-    with pytest.raises(IterationGraphNotFoundError):
-        node._create_graph_engine(index=0, item="item")
-
-
-def test_iteration_node_propagates_non_graph_not_found_errors():
-    runtime_state = _build_runtime_state()
-    node = _build_iteration_node(
-        graph_config={"nodes": [{"id": "start-node"}], "edges": []},
-        runtime_state=runtime_state,
-        start_node_id="start-node",
-    )
-
-    with pytest.raises(ChildEngineBuilderNotConfiguredError):
-        node._create_graph_engine(index=0, item="item")
@@ -239,7 +239,6 @@ def test_image_link_messages_use_tool_file_id_metadata(tool_node: ToolNode):
 def test_tool_node_passes_node_execution_id_when_runtime_accepts_it(tool_node: ToolNode):
    runtime_handle = ToolRuntimeHandle(raw=object())
    tool_node._runtime.get_runtime = MagicMock(return_value=runtime_handle)
-    tool_node.ensure_execution_id = MagicMock(return_value="node-execution-id")

    result = tool_node._get_tool_runtime(
        variable_pool=tool_node.graph_runtime_state.variable_pool,
@@ -18,12 +18,12 @@ from core.workflow.human_input_adapter import (
 )
 from graphon.enums import BuiltinNodeTypes
 from graphon.nodes.base.variable_template_parser import VariableTemplateParser
+from graphon.runtime import VariablePool


 def test_email_delivery_config_helpers_render_and_sanitize_text() -> None:
-    variable_pool = SimpleNamespace(
-        convert_template=lambda body: SimpleNamespace(text=body.replace("{{#node.value#}}", "42"))
-    )
+    variable_pool = VariablePool()
+    variable_pool.add(["node", "value"], "42")

    rendered = EmailDeliveryConfig.render_body_template(
        body="Open {{#url#}} and use {{#node.value#}}",
@@ -59,6 +59,27 @@ def test_dify_hitl_callback_creates_pause_requested_for_new_form() -> None:
    assert params.node_id == "node-1"


+def test_dify_hitl_callback_scopes_form_to_node_execution() -> None:
+    repository = MagicMock(spec=HumanInputFormRepository)
+    repository.get_form.return_value = None
+    repository.create_form.return_value = SimpleNamespace(id="execution-1")
+    callback = DifyHITLCallback(
+        form_repository=repository,
+        node_data=HumanInputNodeData(
+            title="Approval",
+            form_content="Please approve",
+            user_actions=[UserActionConfig(id="approve", title="Approve")],
+        ),
+        execution_id_getter=lambda: "execution-1",
+    )
+
+    callback(_ctx("run-1", "node-1"))
+
+    repository.get_form.assert_called_once_with("node-1", form_id="execution-1")
+    params: FormCreateParams = repository.create_form.call_args.args[0]
+    assert params.form_id == "execution-1"
+
+
 def test_dify_hitl_callback_returns_completed_for_submitted_form() -> None:
    repository = MagicMock(spec=HumanInputFormRepository)
    repository.get_form.return_value = SimpleNamespace(
@@ -324,6 +324,19 @@ class TestDifyNodeFactoryInit:
            graph_runtime_state=sentinel.graph_runtime_state,
        )

+    def test_with_runtime_state_rebinds_factory(self):
+        factory = object.__new__(node_factory.DifyNodeFactory)
+        factory.graph_init_params = sentinel.graph_init_params
+
+        with patch.object(node_factory, "DifyNodeFactory", return_value=sentinel.factory) as factory_cls:
+            rebound = factory.with_runtime_state(sentinel.graph_runtime_state)
+
+        assert rebound is sentinel.factory
+        factory_cls.assert_called_once_with(
+            graph_init_params=sentinel.graph_init_params,
+            graph_runtime_state=sentinel.graph_runtime_state,
+        )
+
    def test_init_builds_default_dependencies(self):
        graph_init_params = SimpleNamespace(run_context={"context": "value"})
        graph_runtime_state = sentinel.graph_runtime_state
@@ -1,5 +1,4 @@
 from collections import UserString
-from contextlib import nullcontext
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch, sentinel

@@ -10,238 +9,20 @@ from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
 from core.workflow import workflow_entry
 from core.workflow.system_variables import default_system_variables
 from graphon.entities.base_node_data import BaseNodeData
-from graphon.enums import NodeType, WorkflowNodeExecutionStatus
+from graphon.enums import NodeType
 from graphon.errors import WorkflowNodeRunFailedError
 from graphon.file import File, FileTransferMethod, FileType
 from graphon.filters import ResponseStreamFilter
-from graphon.graph import Graph
 from graphon.graph_events import GraphRunFailedEvent
-from graphon.model_runtime.entities.llm_entities import LLMMode, LLMUsage
-from graphon.node_events import NodeRunResult
 from graphon.nodes import BuiltinNodeTypes
-from graphon.nodes.base.node import Node
-from graphon.nodes.llm.entities import ContextConfig, LLMNodeData, ModelConfig
-from graphon.nodes.question_classifier.entities import QuestionClassifierNodeData
-from graphon.runtime import ChildGraphNotFoundError, VariablePool
+from graphon.runtime import VariablePool
 from graphon.variables.variables import StringVariable
-from tests.workflow_test_utils import build_test_graph_init_params, build_test_variable_pool


 def _build_typed_node_config(node_type: NodeType):
    return {"id": "node-id", "data": BaseNodeData(type=node_type)}


-def _build_model_config(*, provider: str = "openai", model_name: str = "gpt-4o") -> ModelConfig:
-    return ModelConfig(provider=provider, name=model_name, mode=LLMMode.CHAT)
-
-
-def _build_llm_node_data(*, provider: str = "openai", model_name: str = "gpt-4o") -> LLMNodeData:
-    return LLMNodeData(
-        type=BuiltinNodeTypes.LLM,
-        title="Child Model",
-        model=_build_model_config(provider=provider, model_name=model_name),
-        prompt_template=[],
-        context=ContextConfig(enabled=False),
-    )
-
-
-def _build_question_classifier_node_data(
-    *, provider: str = "openai", model_name: str = "gpt-4o"
-) -> QuestionClassifierNodeData:
-    return QuestionClassifierNodeData(
-        type=BuiltinNodeTypes.QUESTION_CLASSIFIER,
-        title="Child Model",
-        query_variable_selector=["sys", "query"],
-        model=_build_model_config(provider=provider, model_name=model_name),
-        classes=[],
-    )
-
-
-class _FakeModelNodeMixin:
-    @classmethod
-    def version(cls) -> str:
-        return "1"
-
-    def post_init(self) -> None:
-        self.model_instance = SimpleNamespace(provider="stale-provider", model_name="stale-model")
-        self.usage_snapshot = LLMUsage.empty_usage()
-        self.usage_snapshot.total_tokens = 1
-
-    def _run(self) -> NodeRunResult:
-        return NodeRunResult(
-            status=WorkflowNodeExecutionStatus.SUCCEEDED,
-            inputs={
-                "model_provider": self.node_data.model.provider,
-                "model_name": self.node_data.model.name,
-            },
-            llm_usage=self.usage_snapshot,
-        )
-
-
-class _FakeLLMNode(_FakeModelNodeMixin, Node[LLMNodeData]):
-    node_type = BuiltinNodeTypes.LLM
-
-
-class _FakeQuestionClassifierNode(_FakeModelNodeMixin, Node[QuestionClassifierNodeData]):
-    node_type = BuiltinNodeTypes.QUESTION_CLASSIFIER
-
-
-class TestWorkflowChildEngineBuilder:
-    @pytest.mark.parametrize(
-        ("graph_config", "node_id", "expected"),
-        [
-            ({"nodes": [{"id": "root"}]}, "root", True),
-            ({"nodes": [{"id": "root"}]}, "other", False),
-            ({"nodes": "invalid"}, "root", None),
-            ({"nodes": ["invalid"]}, "root", None),
-        ],
-    )
-    def test_has_node_id(self, graph_config, node_id, expected):
-        result = workflow_entry._WorkflowChildEngineBuilder._has_node_id(graph_config, node_id)
-
-        assert result is expected
-
-    def test_build_child_engine_raises_when_root_node_is_missing(self):
-        builder = workflow_entry._WorkflowChildEngineBuilder(tenant_id="tenant-id")
-        graph_init_params = SimpleNamespace(graph_config={"nodes": []})
-        parent_graph_runtime_state = SimpleNamespace(
-            execution_context=sentinel.execution_context,
-            variable_pool=sentinel.variable_pool,
-        )
-
-        with patch.object(workflow_entry, "DifyNodeFactory", return_value=sentinel.factory):
-            with pytest.raises(ChildGraphNotFoundError, match="child graph root node 'missing' not found"):
-                builder.build_child_engine(
-                    workflow_id="workflow-id",
-                    graph_init_params=graph_init_params,
-                    parent_graph_runtime_state=parent_graph_runtime_state,
-                    root_node_id="missing",
-                )
-
-    def test_build_child_engine_constructs_graph_engine_with_quota_layer_only(self):
-        builder = workflow_entry._WorkflowChildEngineBuilder(tenant_id="tenant-id")
-        graph_init_params = SimpleNamespace(graph_config={"nodes": [{"id": "root"}]})
-        parent_graph_runtime_state = SimpleNamespace(
-            execution_context=sentinel.execution_context,
-            variable_pool=sentinel.parent_variable_pool,
-        )
-        child_graph = sentinel.child_graph
-        child_graph_runtime_state = sentinel.child_graph_runtime_state
-        child_engine = MagicMock()
-
-        with (
-            patch.object(workflow_entry.time, "perf_counter", return_value=123.0),
-            patch.object(
-                workflow_entry,
-                "GraphRuntimeState",
-                return_value=child_graph_runtime_state,
-            ) as graph_runtime_state_cls,
-            patch.object(workflow_entry, "DifyNodeFactory", return_value=sentinel.factory) as dify_node_factory,
-            patch.object(workflow_entry.Graph, "init", return_value=child_graph) as graph_init,
-            patch.object(workflow_entry, "GraphEngine", return_value=child_engine) as graph_engine_cls,
-            patch.object(workflow_entry, "GraphEngineConfig", return_value=sentinel.graph_engine_config),
-            patch.object(workflow_entry, "InMemoryChannel", return_value=sentinel.command_channel),
-            patch.object(workflow_entry, "LLMQuotaLayer", return_value=sentinel.llm_quota_layer) as llm_quota_layer_cls,
-        ):
-            result = builder.build_child_engine(
-                workflow_id="workflow-id",
-                graph_init_params=graph_init_params,
-                parent_graph_runtime_state=parent_graph_runtime_state,
-                root_node_id="root",
-                variable_pool=sentinel.child_variable_pool,
-            )
-
-        assert result is child_engine
-        graph_runtime_state_cls.assert_called_once_with(
-            variable_pool=sentinel.child_variable_pool,
-            start_at=123.0,
-            execution_context=sentinel.execution_context,
-        )
-        dify_node_factory.assert_called_once_with(
-            graph_init_params=graph_init_params,
-            graph_runtime_state=child_graph_runtime_state,
-        )
-        graph_init.assert_called_once_with(
-            graph_config={"nodes": [{"id": "root"}]},
-            node_factory=sentinel.factory,
-            root_node_id="root",
-        )
-        graph_engine_cls.assert_called_once_with(
-            workflow_id="workflow-id",
-            graph=child_graph,
-            graph_runtime_state=child_graph_runtime_state,
-            command_channel=sentinel.command_channel,
-            config=sentinel.graph_engine_config,
-            child_engine_builder=builder,
-        )
-        llm_quota_layer_cls.assert_called_once_with(tenant_id="tenant-id")
-        assert child_engine.layer.call_args_list == [((sentinel.llm_quota_layer,), {})]
-
-    @pytest.mark.parametrize("node_cls", [_FakeLLMNode, _FakeQuestionClassifierNode])
-    def test_build_child_engine_runs_llm_quota_layer_for_child_model_nodes(self, node_cls):
-        builder = workflow_entry._WorkflowChildEngineBuilder(tenant_id="tenant-id")
-        graph_init_params = build_test_graph_init_params(
-            graph_config={"nodes": [{"id": "root"}], "edges": []},
-        )
-        parent_graph_runtime_state = SimpleNamespace(
-            execution_context=nullcontext(None),
-            variable_pool=build_test_variable_pool(),
-        )
-        created_node: dict[str, _FakeLLMNode | _FakeQuestionClassifierNode] = {}
-
-        def build_graph(*, graph_config, node_factory, root_node_id):
-            _ = graph_config
-            node_data = _build_llm_node_data() if node_cls is _FakeLLMNode else _build_question_classifier_node_data()
-            node = node_cls(
-                node_id=root_node_id,
-                data=node_data,
-                graph_init_params=node_factory.graph_init_params,
-                graph_runtime_state=node_factory.graph_runtime_state,
-            )
-            created_node["node"] = node
-            return Graph(
-                nodes={root_node_id: node},
-                edges={},
-                in_edges={},
-                out_edges={},
-                root_node=node,
-            )
-
-        with (
-            patch.object(
-                workflow_entry,
-                "DifyNodeFactory",
-                side_effect=lambda graph_init_params, graph_runtime_state: SimpleNamespace(
-                    graph_init_params=graph_init_params,
-                    graph_runtime_state=graph_runtime_state,
-                ),
-            ),
-            patch.object(workflow_entry.Graph, "init", side_effect=build_graph),
-            patch("core.app.workflow.layers.llm_quota.ensure_llm_quota_available_for_model") as ensure_quota,
-            patch("core.app.workflow.layers.llm_quota.deduct_llm_quota_for_model") as deduct_quota,
-        ):
-            child_engine = builder.build_child_engine(
-                workflow_id="workflow-id",
-                graph_init_params=graph_init_params,
-                parent_graph_runtime_state=parent_graph_runtime_state,
-                root_node_id="root",
-            )
-            list(child_engine.run())
-
-        node = created_node["node"]
-        ensure_quota.assert_called_once_with(
-            tenant_id="tenant-id",
-            provider=node.node_data.model.provider,
-            model=node.node_data.model.name,
-        )
-        deduct_quota.assert_called_once_with(
-            tenant_id="tenant-id",
-            provider=node.node_data.model.provider,
-            model=node.node_data.model.name,
-            usage=node.usage_snapshot,
-        )
-
-
 def _build_minimal_workflow_entry(
    monkeypatch: pytest.MonkeyPatch,
    *,
@@ -249,7 +30,7 @@ def _build_minimal_workflow_entry(
 ) -> workflow_entry.WorkflowEntry:
    """Construct a minimal WorkflowEntry with GraphEngine construction mocked out."""
    graph_engine = MagicMock()
-    graph_runtime_state = SimpleNamespace(execution_context=None)
+    graph_runtime_state = SimpleNamespace(_execution_context=None)

    monkeypatch.setattr(workflow_entry, "capture_current_context", lambda: sentinel.execution_context)
    monkeypatch.setattr(workflow_entry, "GraphEngine", MagicMock(return_value=graph_engine))
@@ -294,7 +75,7 @@ class TestWorkflowEntryInit:

    def test_applies_debug_and_observability_layers(self):
        graph_engine = MagicMock()
-        graph_runtime_state = SimpleNamespace(execution_context=None)
+        graph_runtime_state = SimpleNamespace(_execution_context=None)
        debug_layer = sentinel.debug_layer
        execution_limits_layer = sentinel.execution_limits_layer
        llm_quota_layer = sentinel.llm_quota_layer
@@ -339,9 +120,8 @@ class TestWorkflowEntryInit:
            graph_runtime_state=graph_runtime_state,
            command_channel=sentinel.command_channel,
            config=sentinel.graph_engine_config,
-            child_engine_builder=entry._child_engine_builder,
        )
-        assert graph_runtime_state.execution_context is sentinel.execution_context
+        assert graph_runtime_state._execution_context is sentinel.execution_context
        debug_logging_layer.assert_called_once_with(
            level="DEBUG",
            include_inputs=True,
@@ -958,7 +738,7 @@ class TestWorkflowEntryTracing:
        layer = MagicMock()

        class FakeNode:
-            def ensure_execution_id(self):
+            def bind_execution_id(self, _execution_id):
                return None

            def run(self):
@@ -979,7 +759,7 @@ class TestWorkflowEntryTracing:
        layer = MagicMock()

        class FakeNode:
-            def ensure_execution_id(self):
+            def bind_execution_id(self, _execution_id):
                return None

            def run(self):
@@ -1,13 +1,12 @@
-from collections.abc import Generator
-from contextlib import contextmanager
+from collections.abc import Iterator
 from types import SimpleNamespace
 from unittest.mock import ANY, patch
 from uuid import uuid4

 import pytest
-from sqlalchemy import create_engine, select
+from sqlalchemy import select
 from sqlalchemy.engine import Engine
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import Session, sessionmaker

 from core.app.entities.app_invoke_entities import ChatAppGenerateEntity
 from core.entities.provider_entities import ProviderQuotaType, QuotaUnit
@@ -16,40 +15,29 @@ from models import TenantCreditPool
 from models.provider import ProviderType


-@contextmanager
-def _patched_credit_pool_session_factory(engine: Engine) -> Generator[None, None, None]:
-    session_maker = sessionmaker(bind=engine, expire_on_commit=False)
-    sessions = []
-
-    def _session():
-        session = session_maker()
-        sessions.append(session)
-        return session
-
-    with patch("events.event_handlers.update_provider_when_message_created.db", SimpleNamespace(session=_session)):
-        try:
-            yield
-        finally:
-            for session in sessions:
-                session.close()
+@pytest.fixture
+def credit_pool_session_factory(sqlite_engine: Engine) -> Iterator[sessionmaker[Session]]:
+    """Bind message-created accounting to fixture-owned SQLite sessions."""
+    TenantCreditPool.__table__.create(sqlite_engine)
+    session_factory = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
+    with patch("events.event_handlers.update_provider_when_message_created.db.session", session_factory):
+        yield session_factory


-def test_message_created_trial_credit_accounting_does_not_raise_when_balance_is_insufficient() -> None:
-    engine = create_engine("sqlite:///:memory:")
-    TenantCreditPool.__table__.create(engine)
+def test_message_created_trial_credit_accounting_does_not_raise_when_balance_is_insufficient(
+    credit_pool_session_factory: sessionmaker[Session],
+) -> None:
    tenant_id = str(uuid4())
    pool_id = str(uuid4())
-    with engine.begin() as connection:
-        connection.execute(
-            TenantCreditPool.__table__.insert(),
-            {
-                "id": pool_id,
-                "tenant_id": tenant_id,
-                "pool_type": ProviderQuotaType.TRIAL,
-                "quota_limit": 10,
-                "quota_used": 9,
-            },
-        )
+    pool = TenantCreditPool(
+        tenant_id=tenant_id,
+        pool_type=ProviderQuotaType.TRIAL,
+        quota_limit=10,
+        quota_used=9,
+    )
+    pool.id = pool_id
+    with credit_pool_session_factory.begin() as session:
+        session.add(pool)

    system_configuration = SimpleNamespace(
        current_quota_type=ProviderQuotaType.TRIAL,
@@ -77,7 +65,6 @@ def test_message_created_trial_credit_accounting_does_not_raise_when_balance_is_
    message = SimpleNamespace(message_tokens=2, answer_tokens=1)

    with (
-        _patched_credit_pool_session_factory(engine),
        patch.object(update_provider_when_message_created, "_execute_provider_updates"),
    ):
        update_provider_when_message_created.handle(
@@ -85,8 +72,8 @@ def test_message_created_trial_credit_accounting_does_not_raise_when_balance_is_
            application_generate_entity=application_generate_entity,
        )

-    with engine.connect() as connection:
-        quota_used = connection.scalar(select(TenantCreditPool.quota_used).where(TenantCreditPool.id == pool_id))
+    with credit_pool_session_factory() as session:
+        quota_used = session.scalar(select(TenantCreditPool.quota_used).where(TenantCreditPool.id == pool_id))

    assert quota_used == 10

@@ -27,6 +27,7 @@ from libs.broadcast_channel.redis.sharded_channel import (
    ShardedTopic,
    _RedisShardedSubscription,
 )
+from libs.broadcast_channel.signals import SIG_CLOSE


 class TestBroadcastChannel:
@@ -1239,6 +1240,30 @@ class TestRedisSubscriptionCommon:
        subscription_type, _ = subscription_params
        assert subscription._get_subscription_type() == subscription_type

+    def test_listener_ignores_close_signal_from_another_subscription(self, subscription, subscription_params):
+        subscription_type, _ = subscription_params
+        topic = f"test-{subscription_type}-topic"
+        message_type = "message" if subscription_type == "regular" else "smessage"
+        messages = iter(
+            [
+                {"type": message_type, "channel": topic, "data": SIG_CLOSE},
+                {"type": message_type, "channel": topic, "data": b"next-event"},
+            ]
+        )
+
+        def get_message():
+            try:
+                return next(messages)
+            except StopIteration:
+                subscription._closed.set()
+                return None
+
+        subscription._get_message = get_message
+        subscription._listen()
+
+        assert subscription._queue.get_nowait() == b"next-event"
+        assert subscription._queue.empty()
+
    # ==================== Lifecycle Tests ====================

    def test_start_if_needed_first_call(self, subscription, subscription_params, mock_pubsub: MagicMock):
@@ -12,6 +12,7 @@ from libs.broadcast_channel.redis.streams_channel import (
    StreamsTopic,
    _StreamsSubscription,
 )
+from libs.broadcast_channel.signals import SIG_CLOSE


 class FakeStreamsRedis:
@@ -282,6 +283,34 @@ class TestStreamsSubscription:

        assert received == case.expected_messages

+    def test_listener_ignores_close_signal_from_another_subscription(self):
+        class OneShotRedis:
+            def __init__(self) -> None:
+                self._calls = 0
+
+            def xread(self, streams: dict[str, Any], block: int | None = None, count: int | None = None):
+                self._calls += 1
+                if self._calls == 1:
+                    key = next(iter(streams))
+                    return [
+                        (
+                            key,
+                            [
+                                ("1-0", {b"data": SIG_CLOSE}),
+                                ("2-0", {b"data": b"next-event"}),
+                            ],
+                        )
+                    ]
+                subscription._closed = True
+                return []
+
+        subscription = _StreamsSubscription(OneShotRedis(), "stream:close-signal")
+        subscription._listen()
+
+        assert subscription._queue.get_nowait() == b"next-event"
+        assert subscription._queue.get_nowait() is subscription._SENTINEL
+        assert subscription._queue.empty()
+
    def test_iterator_yields_messages_until_subscription_is_closed(self, streams_channel: StreamsBroadcastChannel):
        topic = streams_channel.topic("iter")
        subscription = topic.subscribe()
@@ -5,6 +5,7 @@ from __future__ import annotations
 from unittest.mock import MagicMock, patch

 import pytest
+from sqlalchemy.orm import Session

 from services.agent.skill_tool_inference_service import (
    SkillToolInferenceError,
@@ -29,7 +30,8 @@ def _service(preview=_SKILL_MD_PREVIEW):
    return SkillToolInferenceService(drive_service=drive), drive


-def test_infer_returns_suggestions_with_inferred_from(monkeypatch):
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_infer_returns_suggestions_with_inferred_from(monkeypatch, sqlite_session: Session):
    service, drive = _service()
    raw = (
        '{"inferable": true, "reason": null, "cli_tools": [{"name": "ffmpeg",'
@@ -38,8 +40,12 @@ def test_infer_returns_suggestions_with_inferred_from(monkeypatch):
        ' "env_suggestions": [{"key": "OPENAI_API_KEY", "reason": "whisper call", "secret_likely": true}]}]}'
    )
    with patch.object(SkillToolInferenceService, "_invoke", staticmethod(lambda **kwargs: raw)):
-        session = MagicMock()
-        result = service.infer(tenant_id="t-1", agent_id="a-1", slug="audio-transcribe", session=session)
+        result = service.infer(
+            tenant_id="t-1",
+            agent_id="a-1",
+            slug="audio-transcribe",
+            session=sqlite_session,
+        )

    assert result["inferable"] is True
    tool = result["cli_tools"][0]
@@ -47,11 +53,13 @@ def test_infer_returns_suggestions_with_inferred_from(monkeypatch):
    assert tool["inferred_from"] == "audio-transcribe"
    assert tool["env_suggestions"] == [{"key": "OPENAI_API_KEY", "reason": "whisper call", "secret_likely": True}]
    drive.preview.assert_called_once_with(
-        tenant_id="t-1", agent_id="a-1", key="audio-transcribe/SKILL.md", session=session
+        tenant_id="t-1", agent_id="a-1", key="audio-transcribe/SKILL.md", session=sqlite_session
    )
+    assert not sqlite_session.in_transaction()


-def test_infer_threads_skill_md_into_the_prompt(monkeypatch):
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_infer_threads_skill_md_into_the_prompt(monkeypatch, sqlite_session: Session):
    service, _ = _service()
    captured: dict[str, str] = {}

@@ -60,21 +68,25 @@ def test_infer_threads_skill_md_into_the_prompt(monkeypatch):
        return '{"inferable": false, "cli_tools": [], "reason": "none"}'

    with patch.object(SkillToolInferenceService, "_invoke", staticmethod(fake_invoke)):
-        service.infer(tenant_id="t-1", agent_id="a-1", slug="audio-transcribe", session=MagicMock())
+        service.infer(tenant_id="t-1", agent_id="a-1", slug="audio-transcribe", session=sqlite_session)

    assert "Files inside the skill package" not in captured["prompt"]
    assert "ffmpeg" in captured["prompt"]  # SKILL.md body present
+    assert not sqlite_session.in_transaction()


-def test_infer_not_inferable_passes_reason_through(monkeypatch):
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_infer_not_inferable_passes_reason_through(monkeypatch, sqlite_session: Session):
    service, _ = _service()
    raw = '{"inferable": false, "cli_tools": [], "reason": "SKILL.md 未描述任何外部命令依赖"}'
    with patch.object(SkillToolInferenceService, "_invoke", staticmethod(lambda **kwargs: raw)):
-        result = service.infer(tenant_id="t-1", agent_id="a-1", slug="audio-transcribe", session=MagicMock())
+        result = service.infer(tenant_id="t-1", agent_id="a-1", slug="audio-transcribe", session=sqlite_session)
    assert result == {"inferable": False, "cli_tools": [], "reason": "SKILL.md 未描述任何外部命令依赖"}
+    assert not sqlite_session.in_transaction()


-def test_infer_retries_once_then_422(monkeypatch):
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_infer_retries_once_then_422(monkeypatch, sqlite_session: Session):
    service, _ = _service()
    calls: list[int] = []

@@ -84,37 +96,44 @@ def test_infer_retries_once_then_422(monkeypatch):

    with patch.object(SkillToolInferenceService, "_invoke", staticmethod(bad_invoke)):
        with pytest.raises(SkillToolInferenceError) as exc_info:
-            service.infer(tenant_id="t-1", agent_id="a-1", slug="audio-transcribe", session=MagicMock())
+            service.infer(tenant_id="t-1", agent_id="a-1", slug="audio-transcribe", session=sqlite_session)

    assert len(calls) == 2  # one retry
    assert exc_info.value.code == "inference_failed"
    assert exc_info.value.status_code == 422
+    assert not sqlite_session.in_transaction()


-def test_infer_repairs_slightly_malformed_json(monkeypatch):
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_infer_repairs_slightly_malformed_json(monkeypatch, sqlite_session: Session):
    service, _ = _service()
    raw = 'Here you go: {"inferable": true, "cli_tools": [], "reason": null,}'
    with patch.object(SkillToolInferenceService, "_invoke", staticmethod(lambda **kwargs: raw)):
-        result = service.infer(tenant_id="t-1", agent_id="a-1", slug="audio-transcribe", session=MagicMock())
+        result = service.infer(tenant_id="t-1", agent_id="a-1", slug="audio-transcribe", session=sqlite_session)
    assert result["inferable"] is True
+    assert not sqlite_session.in_transaction()


-def test_missing_skill_maps_to_404():
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_missing_skill_maps_to_404(sqlite_session: Session):
    drive = MagicMock()
    drive.preview.side_effect = AgentDriveError("drive_key_not_found", "nope", status_code=404)
    service = SkillToolInferenceService(drive_service=drive)

    with pytest.raises(SkillToolInferenceError) as exc_info:
-        service.infer(tenant_id="t-1", agent_id="a-1", slug="ghost", session=MagicMock())
+        service.infer(tenant_id="t-1", agent_id="a-1", slug="ghost", session=sqlite_session)
    assert exc_info.value.code == "skill_not_found"
    assert exc_info.value.status_code == 404
+    assert not sqlite_session.in_transaction()


-def test_binary_skill_md_maps_to_404():
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_binary_skill_md_maps_to_404(sqlite_session: Session):
    service, _ = _service(preview={"key": "x/SKILL.md", "size": 1, "truncated": False, "binary": True, "text": None})
    with pytest.raises(SkillToolInferenceError) as exc_info:
-        service.infer(tenant_id="t-1", agent_id="a-1", slug="x", session=MagicMock())
+        service.infer(tenant_id="t-1", agent_id="a-1", slug="x", session=sqlite_session)
    assert exc_info.value.code == "skill_not_found"
+    assert not sqlite_session.in_transaction()


 # ── real-path coverage: _invoke / passthrough ────────────────────────────────
@@ -157,11 +176,13 @@ def test_invoke_maps_model_failure_to_422_and_success_returns_text(monkeypatch):
    assert call["stream"] is False


-def test_load_skill_md_passes_through_non_missing_drive_errors():
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_load_skill_md_passes_through_non_missing_drive_errors(sqlite_session: Session):
    drive = MagicMock()
    drive.preview.side_effect = AgentDriveError("agent_not_found", "tenant mismatch", status_code=404)
    service = SkillToolInferenceService(drive_service=drive)

    with pytest.raises(SkillToolInferenceError) as exc_info:
-        service.infer(tenant_id="t-1", agent_id="a-1", slug="x", session=MagicMock())
+        service.infer(tenant_id="t-1", agent_id="a-1", slug="x", session=sqlite_session)
    assert exc_info.value.code == "agent_not_found"
+    assert not sqlite_session.in_transaction()
@@ -6,6 +6,7 @@ import datetime
 from unittest.mock import MagicMock, patch

 import pytest
+from sqlalchemy.orm import Session

 from repositories.api_workflow_run_repository import WorkflowRunCleanupRef
 from services.retention.workflow_run.clear_free_plan_expired_workflow_run_logs import WorkflowRunCleanup
@@ -472,24 +473,24 @@ class TestRunDryRunMode:


 class TestTriggerLogMethods:
-    def test_delete_trigger_logs(self, cleanup):
-        session = MagicMock()
+    @pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+    def test_delete_trigger_logs(self, cleanup, sqlite_session: Session):
        with patch(
            "services.retention.workflow_run.clear_free_plan_expired_workflow_run_logs.SQLAlchemyWorkflowTriggerLogRepository"
        ) as RepoClass:
            instance = RepoClass.return_value
            instance.delete_by_run_ids.return_value = 5
-            result = cleanup._delete_trigger_logs(session, ["r1", "r2"])
+            result = cleanup._delete_trigger_logs(sqlite_session, ["r1", "r2"])
        assert result == 5

-    def test_count_trigger_logs(self, cleanup):
-        session = MagicMock()
+    @pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+    def test_count_trigger_logs(self, cleanup, sqlite_session: Session):
        with patch(
            "services.retention.workflow_run.clear_free_plan_expired_workflow_run_logs.SQLAlchemyWorkflowTriggerLogRepository"
        ) as RepoClass:
            instance = RepoClass.return_value
            instance.count_by_run_ids.return_value = 3
-            result = cleanup._count_trigger_logs(session, ["r1"])
+            result = cleanup._count_trigger_logs(sqlite_session, ["r1"])
        assert result == 3


@@ -499,26 +500,22 @@ class TestTriggerLogMethods:


 class TestNodeExecutionMethods:
-    def test_count_node_executions(self, cleanup):
-        session = MagicMock()
-        session.get_bind.return_value = MagicMock()
+    @pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+    def test_count_node_executions(self, cleanup, sqlite_session: Session):
        with patch(
            "services.retention.workflow_run.clear_free_plan_expired_workflow_run_logs.DifyAPIRepositoryFactory"
        ) as factory:
            repo = factory.create_api_workflow_node_execution_repository.return_value
            repo.count_by_runs.return_value = (10, 2)
-            with patch("services.retention.workflow_run.clear_free_plan_expired_workflow_run_logs.sessionmaker"):
-                result = cleanup._count_node_executions_by_run_ids(session, ["r1"])
+            result = cleanup._count_node_executions_by_run_ids(sqlite_session, ["r1"])
        assert result == (10, 2)

-    def test_delete_node_executions(self, cleanup):
-        session = MagicMock()
-        session.get_bind.return_value = MagicMock()
+    @pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+    def test_delete_node_executions(self, cleanup, sqlite_session: Session):
        with patch(
            "services.retention.workflow_run.clear_free_plan_expired_workflow_run_logs.DifyAPIRepositoryFactory"
        ) as factory:
            repo = factory.create_api_workflow_node_execution_repository.return_value
            repo.delete_by_runs.return_value = (5, 1)
-            with patch("services.retention.workflow_run.clear_free_plan_expired_workflow_run_logs.sessionmaker"):
-                result = cleanup._delete_node_executions_by_run_ids(session, ["r1"])
+            result = cleanup._delete_node_executions_by_run_ids(sqlite_session, ["r1"])
        assert result == (5, 1)
@@ -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,12 @@ 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
@@ -0,0 +1,238 @@
+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["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 == {
+        "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["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["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
@@ -137,8 +137,7 @@ def _build_resumption_context(task_id: str, *, select_options: list[str] | None
    runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=0.0)
    if select_options is not None:
        runtime_state.variable_pool.add(("start", "options"), select_options)
-    runtime_state.register_paused_node("node-1")
-    runtime_state.outputs = {"result": "value"}
+    runtime_state.set_output("result", "value")
    wrapper = _WorkflowGenerateEntityWrapper(entity=generate_entity)
    return WorkflowResumptionContext(
        generate_entity=wrapper,
@@ -243,7 +242,7 @@ def _build_resumption_context_additional(task_id: str) -> WorkflowResumptionCont
        workflow_execution_id="run-1",
    )
    runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=0.0)
-    runtime_state.outputs = {"answer": "ok"}
+    runtime_state.set_output("answer", "ok")
    wrapper = _WorkflowGenerateEntityWrapper(entity=generate_entity)
    return WorkflowResumptionContext(
        generate_entity=wrapper,
@@ -68,7 +68,7 @@ def _build_resumption_context(task_id: str) -> WorkflowResumptionContext:
        workflow_execution_id="run-1",
    )
    runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=0.0)
-    runtime_state.outputs = {"answer": "ok"}
+    runtime_state.set_output("answer", "ok")
    wrapper = _WorkflowGenerateEntityWrapper(entity=generate_entity)
    return WorkflowResumptionContext(
        generate_entity=wrapper,
@@ -1638,7 +1638,7 @@ requires-dist = [
    { name = "gmpy2", specifier = ">=2.3.0,<3.0.0" },
    { name = "google-api-python-client", specifier = ">=2.198.0,<3.0.0" },
    { name = "google-cloud-aiplatform", specifier = ">=1.160.0,<2.0.0" },
-    { name = "graphon", specifier = "==0.6.0" },
+    { name = "graphon", git = "https://github.com/langgenius/graphon?rev=d48c36fb02d8aa0d31dc6a9140a27c04a370600f" },
    { name = "gunicorn", specifier = ">=26.0.0,<27.0.0" },
    { name = "httpx", extras = ["socks"], specifier = "==0.28.1" },
    { name = "httpx-sse", specifier = "==0.4.3" },
@@ -2991,8 +2991,8 @@ httpx = [

 [[package]]
 name = "graphon"
-version = "0.6.0"
-source = { registry = "https://pypi.org/simple" }
+version = "0.7.0"
+source = { git = "https://github.com/langgenius/graphon?rev=d48c36fb02d8aa0d31dc6a9140a27c04a370600f#d48c36fb02d8aa0d31dc6a9140a27c04a370600f" }
 dependencies = [
    { name = "charset-normalizer" },
    { name = "httpx" },
@@ -3013,10 +3013,6 @@ dependencies = [
    { name = "unstructured", extra = ["docx", "epub", "md", "ppt", "pptx"] },
    { name = "webvtt-py" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/ee/6c/9ea051ed30dc3306e9e77c4486b5a2e5462af45e35daf230d9ec886eb07e/graphon-0.6.0.tar.gz", hash = "sha256:2d3a386899dc7ab8e9767ab96c694ff7e6eb454c045a1e801505cba9c615160d", size = 264404, upload-time = "2026-06-29T15:26:27.437Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/32/36/5b0ece2d61fa091f7d74aea5a48f9cf927202099e8d22b856e0319218e9d/graphon-0.6.0-py3-none-any.whl", hash = "sha256:f1445ccef40c0d0eb50a60af85c1028b26d2d782f357315047acee816acbcb33", size = 376038, upload-time = "2026-06-29T15:26:26.186Z" },
-]

 [[package]]
 name = "graphql-core"
@@ -192,7 +192,7 @@ WORKFLOW_GENERATION_TIMEOUT_MS=180000
 WORKFLOW_FILE_UPLOAD_LIMIT=10
 GRAPH_ENGINE_MIN_WORKERS=3
 GRAPH_ENGINE_MAX_WORKERS=10
-GRAPH_ENGINE_SCALE_UP_THRESHOLD=3
+GRAPH_ENGINE_SCALE_UP_THRESHOLD=0
 GRAPH_ENGINE_SCALE_DOWN_IDLE_TIME=5.0
 ALIYUN_SLS_ACCESS_KEY_ID=
 ALIYUN_SLS_ACCESS_KEY_SECRET=
@@ -807,11 +807,6 @@
      "count": 1
    }
  },
-  "web/app/components/apps/starred-app-card.tsx": {
-    "jsx_a11y/no-noninteractive-element-to-interactive-role": {
-      "count": 1
-    }
-  },
  "web/app/components/base/action-button/index.tsx": {
    "erasable-syntax-only/enums": {
      "count": 1
@@ -3936,14 +3931,6 @@
      "count": 3
    }
  },
-  "web/app/components/tools/tool-provider-grid.tsx": {
-    "jsx_a11y/click-events-have-key-events": {
-      "count": 1
-    },
-    "jsx_a11y/no-static-element-interactions": {
-      "count": 1
-    }
-  },
  "web/app/components/tools/types.ts": {
    "erasable-syntax-only/enums": {
      "count": 4
@@ -5702,11 +5689,6 @@
      "count": 1
    }
  },
-  "web/app/components/workflow/panel/human-input-form-list.tsx": {
-    "typescript/no-explicit-any": {
-      "count": 1
-    }
-  },
  "web/app/components/workflow/panel/inputs-panel.tsx": {
    "jsx_a11y/no-autofocus": {
      "count": 1
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
-LAN-	2343b3f072	fix(workflow): align with GraphOn 0.7.0 Update the GraphOn PR pin and lockfile to its latest frame-aware runtime. Align worker scaling and container event typing with the new pause/resume scheduler while keeping Dify's Human Input adapters compatible.	2026-07-28 13:18:35 +08:00
-LAN-	d1b096e32a	chore: merge main into human input branch Resolve the workflow stream handler test conflict while preserving resumable Human Input coverage and the latest timeout assertions from main.	2026-07-21 00:08:03 +08:00
yyhandGitHub	b7193d1cba	refactor(web): remove basePath from hydration boundary (#39308 )	2026-07-20 14:46:37 +00:00
wangxiaolei GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	ef0115d340	chore: when dataset permission is not all team, update rbac config to… (#39227 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-20 12:59:29 +00:00
Checo GitHub yyh	3afb9b3230	fix(web): stop doubling basePath in auth refresh redirects (#39273 ) Co-authored-by: yyh <yuanyouhuilyz@gmail.com>	2026-07-20 12:22:40 +00:00
JoelandGitHub	0862641533	feat: support export agent dsl in sidebar (#39299 )	2026-07-20 10:41:26 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	cbd7872520	test: use SQLite sessions in unit misc (#39121 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-20 10:38:52 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	23882a704e	test: use SQLite sessions in unit misc (#39122 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-20 10:38:49 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	1bd5254ea7	test: use SQLite sessions in unit misc (#39123 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-20 10:38:46 +00:00
yyhandGitHub	e13069dbe3	revert(ci): restore default Depot runners (#39297 )	2026-07-20 10:18:52 +00:00
github-actions[bot]GitHubclaude[bot] <41898282+claude[bot]@users.noreply.github.com>yyh	edd3d4a218	chore(i18n): sync translations with en-US (#39296 ) Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com> Co-authored-by: yyh <92089059+lyzno1@users.noreply.github.com>	2026-07-20 10:15:37 +00:00
JoelandGitHub	79751f7622	chore: display non-LLM settings in integration tool details (#39295 )	2026-07-20 09:52:54 +00:00
Jingyi GitHub hjlarryautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>yyh yyh	cb64446fa3	feat(web): add step-by-step tour shell (#38785 ) Co-authored-by: hjlarry <hjlarry@163.com> Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: yyh <yuanyouhuilyz@gmail.com> Co-authored-by: yyh <92089059+lyzno1@users.noreply.github.com>	2026-07-20 09:23:25 +00:00
JoelandGitHub	e77ee82526	fix: show persistent error when workflow draft save fails (#39293 )	2026-07-20 08:48:38 +00:00
yyhandGitHub	b3f163fb0f	chore(deps): update pnpm and workspace dependencies (#39292 )	2026-07-20 08:43:21 +00:00
autofix-ci[bot]andGitHub	cf345f2dcb	[autofix.ci] apply automated fixes	2026-07-20 02:30:27 +00:00
-LAN-	1381334b8c	fix(chat): synchronize human input resume streams Gate Human Input submissions until the continuation SSE has observed the active workflow pause. Preserve paused Advanced Chat conversations across refreshes, isolate stale run callbacks, and cover published and installed chat submission paths.	2026-07-20 10:26:22 +08:00
-LAN-	a37c82fb50	fix(workflow): align human input resume streams Use one snapshot-backed continuation stream for Debug and Installed App workflows, and let the Console events endpoint keep that stream open across pauses.	2026-07-20 09:09:53 +08:00
-LAN-	373498c406	feat(workflow): support human input in loop and iteration Update the GraphOn dependency and Dify runtime, persistence, event streaming, and UI integration so nested Human Input forms pause and resume correctly.	2026-07-20 08:28:14 +08:00